[BAHIR-255] Kudu flink-sql add hash partitions nums config&onwer config

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: Flink-1.1.0
• Fix Version/s: Flink-1.1.0
• Component/s: Flink Streaming Connectors
• Labels: flink-kudu hash-partition-nums
• Environment: maxos
• Flags: Patch
• Docs Text: Thank you for your review
• Assignee: Unassigned
• Reporter: ShiminHuang
• Created: 25/Dec/20 09:10
• Updated: 19/May/22 11:56
• Resolved: 25/Dec/20 09:25

Description

Hash Partition Feature

• The default number of buckets in the previous hash partition was 2* replicas, and there were some problems in the production environment, so I felt I should provide the displayed hash partition configuration.
• Add the hash partition number configuration 'kudu.hash-partition-nums',which defaults to 3 if not specified.

Singleton Test

 

public class NewTablePropertiesTest extends KuduCatalogTest{
    private KuduCatalog catalog;
    private StreamTableEnvironment tableEnv;

    @BeforeEach
    public void init() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        catalog = new KuduCatalog(harness.getMasterAddressesAsString());
        tableEnv = KuduTableTestUtils.createTableEnvWithBlinkPlannerStreamingMode(env);
        tableEnv.registerCatalog("kudu", catalog);
        tableEnv.useCatalog("kudu");
    }

    @Test
    public void testHashPartitionNums() throws TableNotExistException {
        catalog.dropTable(ObjectPath.fromString(EnvironmentSettings.DEFAULT_BUILTIN_DATABASE + ".TestTable1"), true);
        catalog.dropTable(ObjectPath.fromString(EnvironmentSettings.DEFAULT_BUILTIN_DATABASE + ".TestTable2"), true);
        tableEnv.executeSql("CREATE TABLE TestTable1 (`first` STRING, `second` String) WITH ('kudu.hash-columns' = 'first', 'kudu.primary-key-columns' = 'first')");
        tableEnv.executeSql("CREATE TABLE TestTable2 (`first` STRING, `second` String) WITH ('kudu.hash-columns' = 'first','kudu.hash-partition-nums'='6', 'kudu.primary-key-columns' = 'first')");
    }


    @Test
    public void testOwner() throws TableNotExistException {
        catalog.dropTable(ObjectPath.fromString(EnvironmentSettings.DEFAULT_BUILTIN_DATABASE + ".TestTable1"), true);
        tableEnv.executeSql("CREATE TABLE TestTable1 (`first` STRING, `second` String) WITH ('kudu.hash-columns' = 'first', 'kudu.primary-key-columns' = 'first','kudu.table-owner'='admin')");
    }
}

Other Ideas

• Regarding the range partition, I think a JSON List can be passed in to describe the corresponding range partition, and I think the hash partition and range partition must exist in one

 

 

Issue Links

• links to: GitHub Pull Request #103