cshuo commented on code in PR #19540:
URL: https://github.com/apache/hudi/pull/19540#discussion_r3735355435
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java:
##########
@@ -273,4 +294,263 @@ void testTableExist() throws IOException {
assertTrue(StreamerUtil.tableExists(basePath,
HadoopConfigurations.getHadoopConf(conf)));
}
}
+
+ @Test
+ void testBuildProperties() {
+ TypedProperties properties = StreamerUtil.buildProperties(
+ Arrays.asList("hoodie.test.one=1", "hoodie.test.two=two"));
+
+ assertEquals("1", properties.getString("hoodie.test.one"));
+ assertEquals("two", properties.getString("hoodie.test.two"));
+ assertThrows(IllegalArgumentException.class,
+ () ->
StreamerUtil.buildProperties(Collections.singletonList("invalid")));
+ }
+
+ @Test
+ void testSourceSchemaConfiguration() {
+ Configuration conf = new Configuration();
+ String schema =
"{\"type\":\"record\",\"name\":\"record\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"}]}";
+ conf.set(FlinkOptions.SOURCE_AVRO_SCHEMA, schema);
+
+ HoodieSchema sourceSchema = StreamerUtil.getSourceSchema(conf);
+
+ assertTrue(sourceSchema.getField("id").isPresent());
+ assertThrows(HoodieException.class,
+ () -> StreamerUtil.getSourceSchema(new Configuration()));
+ }
+
+ @Test
+ void testConfigurationConversions() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.COMPACTION_MAX_MEMORY, 256);
+ conf.set(FlinkOptions.ORDERING_FIELDS, "ts");
+ conf.set(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BLOOM.name());
+
+ TypedProperties properties = StreamerUtil.flinkConf2TypedProperties(conf);
+
+ assertEquals(conf.get(FlinkOptions.TABLE_TYPE),
+ properties.getString(HoodieTableConfig.TYPE.key()));
+ assertEquals(256L * 1024 * 1024,
StreamerUtil.getMaxCompactionMemoryInBytes(conf));
+ assertEquals("ts",
StreamerUtil.getPayloadConfig(conf).getString(HoodiePayloadConfig.ORDERING_FIELDS));
+ assertEquals(HoodieIndex.IndexType.BLOOM.name(),
+
StreamerUtil.getIndexConfig(conf).getString(HoodieIndexConfig.INDEX_TYPE));
+ }
+
+ @Test
+ void testSimplePathAndFileUtilities() {
+ assertEquals("partition_file-id",
StreamerUtil.generateBucketKey("partition", "file-id"));
+
+ assertFalse(StreamerUtil.isValidFile(pathInfo("file.parquet", 4)));
+ assertTrue(StreamerUtil.isValidFile(pathInfo("file.parquet", 5)));
+ assertFalse(StreamerUtil.isValidFile(pathInfo("file.log", 6)));
+ assertTrue(StreamerUtil.isValidFile(pathInfo("file.log", 7)));
+ assertFalse(StreamerUtil.isValidFile(pathInfo("file.orc", 3)));
+ assertTrue(StreamerUtil.isValidFile(pathInfo("file.orc", 4)));
+ assertFalse(StreamerUtil.isValidFile(pathInfo("file.unknown", 0)));
+ assertTrue(StreamerUtil.isValidFile(pathInfo("file.unknown", 1)));
+ }
+
+ @Test
+ void testPartitionExists() throws IOException {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ org.apache.hadoop.conf.Configuration hadoopConf =
HadoopConfigurations.getHadoopConf(conf);
+ assertFalse(StreamerUtil.partitionExists(tempFile.getAbsolutePath(),
"dt=2026-08-06", hadoopConf));
+
+ try (FileSystem fs = HadoopFSUtils.getFs(tempFile.getAbsolutePath(),
hadoopConf)) {
Review Comment:
Fixed in a9caf6499ce. The test now creates the partition with its local
temporary directory and no longer obtains or closes a cached Hadoop FileSystem.
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestDataTypeUtils.java:
##########
@@ -81,4 +95,126 @@ public void testToHoodieSchema() {
assertEquals(HoodieSchemaType.STRING,
requiredSchema.getField("missing").get().schema().getNonNullType().getType());
}
+
+ @Test
+ void testTypePredicatesAndPrecision() {
+ assertTrue(DataTypeUtils.isTimestampType(DataTypes.TIMESTAMP(3)));
+ assertFalse(DataTypeUtils.isTimestampType(DataTypes.TIMESTAMP_LTZ(3)));
+ assertTrue(DataTypeUtils.isDateType(DataTypes.DATE()));
+ assertTrue(DataTypeUtils.isDatetimeType(DataTypes.DATE()));
+ assertTrue(DataTypeUtils.isDatetimeType(DataTypes.TIMESTAMP(3)));
+ assertFalse(DataTypeUtils.isDatetimeType(DataTypes.STRING()));
+ assertEquals(3,
DataTypeUtils.precision(DataTypes.TIMESTAMP(3).getLogicalType()));
+ assertEquals(6,
DataTypeUtils.precision(DataTypes.TIMESTAMP_LTZ(6).getLogicalType()));
+ assertThrows(AssertionError.class,
+ () -> DataTypeUtils.precision(DataTypes.STRING().getLogicalType()));
+ assertTrue(DataTypeUtils.isFamily(
+ DataTypes.INT().getLogicalType(), LogicalTypeFamily.NUMERIC));
+ }
+
+ @Test
+ void testRowTypeProjectionUtilities() {
+ Schema schema = Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .columnByExpression("computed", "id + 1")
+ .build();
+ RowType rowType = DataTypeUtils.toRowType(schema);
+
+ assertEquals(Arrays.asList("id", "name"), rowType.getFieldNames());
+ RowType projected = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("id", DataTypes.INT()))
+ .getLogicalType();
+ assertArrayEquals(new int[] {1, 0}, DataTypeUtils.projectOrdinals(rowType,
projected));
+ assertEquals(Arrays.asList("name", "id"), Arrays.asList(
+ DataTypeUtils.projectRowFields(rowType, new String[] {"name",
"id"})[0].getName(),
+ DataTypeUtils.projectRowFields(rowType, new String[] {"name",
"id"})[1].getName()));
+ }
Review Comment:
Fixed in ea0c0f5cbb1. The projected RowField array is now computed once and
reused for both assertions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
