yihua commented on code in PR #13711:
URL: https://github.com/apache/hudi/pull/13711#discussion_r2353888310
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -670,6 +676,265 @@ public void testSchemaEvolution(String tableType, boolean
useUserProvidedSchema,
defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
}
+ @Test
+ public void testTimestampMillis() throws Exception {
+ // if (zookeeperTestService != null) {
+ // zookeeperTestService.stop();
+ // zookeeperTestService = null;
+ // }
Review Comment:
nit: remove unused code?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -670,6 +676,265 @@ public void testSchemaEvolution(String tableType, boolean
useUserProvidedSchema,
defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
}
+ @Test
+ public void testTimestampMillis() throws Exception {
+ // if (zookeeperTestService != null) {
+ // zookeeperTestService.stop();
+ // zookeeperTestService = null;
+ // }
+ String tableBasePath = basePath + "/testTimestampMillis";
+ defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
+ // Insert data produced with Schema A, pass Schema A
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT,
Collections.singletonList(TestIdentityTransformer.class.getName()),
+ PROPS_FILENAME_TEST_SOURCE, false, true, false, null,
HoodieTableType.MERGE_ON_READ.name());
+ cfg.payloadClassName = DefaultHoodieRecordPayload.class.getName();
+ cfg.recordMergeStrategyId =
HoodieRecordMerger.EVENT_TIME_BASED_MERGE_STRATEGY_UUID;
+ cfg.recordMergeMode = RecordMergeMode.EVENT_TIME_ORDERING;
+ cfg.configs.add("hoodie.streamer.schemaprovider.source.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ cfg.configs.add("hoodie.streamer.schemaprovider.target.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ cfg.configs.add(String.format("%s=%s",
HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key(), "0"));
+ cfg.configs.add("hoodie.datasource.write.row.writer.enable=false");
+
+
+ new HoodieDeltaStreamer(cfg, jsc).sync();
+ assertUseV2Checkpoint(HoodieTestUtils.createMetaClient(storage,
tableBasePath));
+ assertRecordCount(1000, tableBasePath, sqlContext);
+ TestHelpers.assertCommitMetadata("00000", tableBasePath, 1);
+ TableSchemaResolver tableSchemaResolver = new TableSchemaResolver(
+ HoodieTestUtils.createMetaClient(storage, tableBasePath));
+ Schema tableSchema = tableSchemaResolver.getTableAvroSchema(false);
+ assertEquals("timestamp-millis",
tableSchema.getField("current_ts").schema().getLogicalType().getName());
+ assertEquals(1000,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
> '1980-01-01'").count());
+
+ cfg = TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
Collections.singletonList(TestIdentityTransformer.class.getName()),
+ PROPS_FILENAME_TEST_SOURCE, false, true, false, null,
HoodieTableType.MERGE_ON_READ.name());
+ cfg.payloadClassName = DefaultHoodieRecordPayload.class.getName();
+ cfg.recordMergeStrategyId =
HoodieRecordMerger.EVENT_TIME_BASED_MERGE_STRATEGY_UUID;
+ cfg.recordMergeMode = RecordMergeMode.EVENT_TIME_ORDERING;
+ cfg.configs.add("hoodie.streamer.schemaprovider.source.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ cfg.configs.add("hoodie.streamer.schemaprovider.target.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ cfg.configs.add(String.format("%s=%s",
HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key(), "0"));
+ cfg.configs.add("hoodie.datasource.write.row.writer.enable=false");
+
+ new HoodieDeltaStreamer(cfg, jsc).sync();
+ assertUseV2Checkpoint(HoodieTestUtils.createMetaClient(storage,
tableBasePath));
+ assertRecordCount(1450, tableBasePath, sqlContext);
+ TestHelpers.assertCommitMetadata("00001", tableBasePath, 2);
+ tableSchemaResolver = new TableSchemaResolver(
+ HoodieTestUtils.createMetaClient(storage, tableBasePath));
+ tableSchema = tableSchemaResolver.getTableAvroSchema(false);
+ assertEquals("timestamp-millis",
tableSchema.getField("current_ts").schema().getLogicalType().getName());
+ sqlContext.clearCache();
+ assertEquals(1450,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
> '1980-01-01'").count());
+ assertEquals(1450,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
< '2080-01-01'").count());
+ assertEquals(0,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
< '1980-01-01'").count());
+ }
+
+ @Test
+ public void testLogicalTypes() throws Exception {
+ if (zookeeperTestService != null) {
+ zookeeperTestService.stop();
+ zookeeperTestService = null;
+ }
+ try {
+ String tableBasePath = basePath + "/testTimestampMillis";
+ defaultSchemaProviderClassName =
TestHoodieDeltaStreamerSchemaEvolutionBase.TestSchemaProvider.class.getName();
+
+ if (HoodieSparkUtils.isSpark3_3()) {
+
TestHoodieDeltaStreamerSchemaEvolutionBase.TestSchemaProvider.sourceSchema =
HoodieTestDataGenerator.AVRO_TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS;
+
TestHoodieDeltaStreamerSchemaEvolutionBase.TestSchemaProvider.targetSchema =
HoodieTestDataGenerator.AVRO_TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS;
+ AbstractBaseTestSource.schemaStr =
HoodieTestDataGenerator.TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS;
+ AbstractBaseTestSource.avroSchema =
HoodieTestDataGenerator.AVRO_TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS;
+ } else {
+
TestHoodieDeltaStreamerSchemaEvolutionBase.TestSchemaProvider.sourceSchema =
HoodieTestDataGenerator.AVRO_TRIP_LOGICAL_TYPES_SCHEMA;
+
TestHoodieDeltaStreamerSchemaEvolutionBase.TestSchemaProvider.targetSchema =
HoodieTestDataGenerator.AVRO_TRIP_LOGICAL_TYPES_SCHEMA;
+ AbstractBaseTestSource.schemaStr =
HoodieTestDataGenerator.TRIP_LOGICAL_TYPES_SCHEMA;
+ AbstractBaseTestSource.avroSchema =
HoodieTestDataGenerator.AVRO_TRIP_LOGICAL_TYPES_SCHEMA;
+ }
+
+ // Insert data produced with Schema A, pass Schema A
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT,
Collections.singletonList(TestIdentityTransformer.class.getName()),
+ PROPS_FILENAME_TEST_SOURCE, false, true, false, null,
HoodieTableType.MERGE_ON_READ.name());
+ cfg.payloadClassName = DefaultHoodieRecordPayload.class.getName();
+ cfg.recordMergeStrategyId =
HoodieRecordMerger.EVENT_TIME_BASED_MERGE_STRATEGY_UUID;
+ cfg.recordMergeMode = RecordMergeMode.EVENT_TIME_ORDERING;
+ cfg.configs.add(String.format("%s=%s",
HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key(), "0"));
+ cfg.configs.add("hoodie.datasource.write.row.writer.enable=false");
+
+ new HoodieDeltaStreamer(cfg, jsc).sync();
+ assertUseV2Checkpoint(HoodieTestUtils.createMetaClient(storage,
tableBasePath));
+ assertRecordCount(1000, tableBasePath, sqlContext);
+ TestHelpers.assertCommitMetadata("00000", tableBasePath, 1);
+ TableSchemaResolver tableSchemaResolver = new TableSchemaResolver(
+ HoodieTestUtils.createMetaClient(storage, tableBasePath));
+ Schema tableSchema = tableSchemaResolver.getTableAvroSchema(false);
+ Map<String, String> hudiOpts = new HashMap<>();
+ hudiOpts.put("hoodie.datasource.write.recordkey.field", "id");
+ logicalAssertions(tableSchema, tableBasePath, hudiOpts,
HoodieTableVersion.current().versionCode());
+
+
+ cfg = TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
Collections.singletonList(TestIdentityTransformer.class.getName()),
+ PROPS_FILENAME_TEST_SOURCE, false, true, false, null,
HoodieTableType.MERGE_ON_READ.name());
+ cfg.payloadClassName = DefaultHoodieRecordPayload.class.getName();
+ cfg.recordMergeStrategyId =
HoodieRecordMerger.EVENT_TIME_BASED_MERGE_STRATEGY_UUID;
+ cfg.recordMergeMode = RecordMergeMode.EVENT_TIME_ORDERING;
+ cfg.configs.add(String.format("%s=%s",
HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key(), "0"));
+ cfg.configs.add("hoodie.datasource.write.row.writer.enable=false");
+
+ new HoodieDeltaStreamer(cfg, jsc).sync();
+ assertUseV2Checkpoint(HoodieTestUtils.createMetaClient(storage,
tableBasePath));
+ assertRecordCount(1450, tableBasePath, sqlContext);
+ TestHelpers.assertCommitMetadata("00001", tableBasePath, 2);
+ tableSchemaResolver = new TableSchemaResolver(
+ HoodieTestUtils.createMetaClient(storage, tableBasePath));
+ tableSchema = tableSchemaResolver.getTableAvroSchema(false);
+ logicalAssertions(tableSchema, tableBasePath, hudiOpts,
HoodieTableVersion.current().versionCode());
+ } finally {
+ defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
+ AbstractBaseTestSource.schemaStr =
HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+ AbstractBaseTestSource.avroSchema = HoodieTestDataGenerator.AVRO_SCHEMA;
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "EIGHT"})
+ public void testBackwardsCompatibility(HoodieTableVersion version) throws
Exception {
+ TestMercifulJsonToRowConverter.timestampNTZCompatibility(() -> {
+ if (zookeeperTestService != null) {
+ zookeeperTestService.stop();
+ zookeeperTestService = null;
+ }
+ String dirName = "colstats-upgrade-test-v" + version.versionCode();
+ String dataPath = basePath + "/" + dirName;
+ java.nio.file.Path zipOutput = Paths.get(new URI(dataPath));
+ HoodieTestUtils.extractZipToDirectory("col-stats/" + dirName + ".zip",
zipOutput, getClass());
+ String tableBasePath =
zipOutput.resolve("trips_logical_types_json").toString();
+
+ TableSchemaResolver tableSchemaResolver = new TableSchemaResolver(
+ HoodieTestUtils.createMetaClient(storage, tableBasePath));
+ Schema tableSchema = tableSchemaResolver.getTableAvroSchema(false);
+ Map<String, String> hudiOpts = new HashMap<>();
+ hudiOpts.put("hoodie.datasource.write.recordkey.field", "id");
+ logicalAssertions(tableSchema, tableBasePath, hudiOpts,
version.versionCode());
+
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.UPSERT, Collections.emptyList(),
+ "placeholder", false, true, false, null,
HoodieTableType.MERGE_ON_READ.name());
+ cfg.propsFilePath = zipOutput + "/hudi.properties";
+ cfg.schemaProviderClassName =
"org.apache.hudi.utilities.schema.FilebasedSchemaProvider";
+ cfg.sourceOrderingFields = "timestamp";
+ cfg.sourceClassName = "org.apache.hudi.utilities.sources.JsonDFSSource";
+ cfg.targetTableName = "trips_logical_types_json";
+ cfg.configs.add("hoodie.streamer.source.dfs.root=" + zipOutput +
"/data/data_6/");
+ cfg.configs.add(String.format(("%s=%s"),
HoodieWriteConfig.WRITE_TABLE_VERSION.key(), version.versionCode()));
+ cfg.configs.add(String.format(("%s=%s"),
HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(), "100"));
+ cfg.forceDisableCompaction = true;
+ cfg.sourceLimit = 100_000;
+ cfg.ignoreCheckpoint = "12345";
+ new HoodieDeltaStreamer(cfg, jsc).sync();
+ logicalAssertions(tableSchema, tableBasePath, hudiOpts,
version.versionCode());
+ });
+ }
+
+ private void logicalAssertions(Schema tableSchema, String tableBasePath,
Map<String, String> hudiOpts, int tableVersion) {
+ if (tableVersion > 8) {
+ assertEquals("timestamp-millis",
tableSchema.getField("ts_millis").schema().getLogicalType().getName());
+ }
+ assertEquals("timestamp-micros",
tableSchema.getField("ts_micros").schema().getLogicalType().getName());
+ if (tableVersion > 8 && !HoodieSparkUtils.isSpark3_3()) {
+ assertEquals("local-timestamp-millis",
tableSchema.getField("local_ts_millis").schema().getLogicalType().getName());
+ assertEquals("local-timestamp-micros",
tableSchema.getField("local_ts_micros").schema().getLogicalType().getName());
+ }
+
+ assertEquals("date",
tableSchema.getField("event_date").schema().getLogicalType().getName());
+
+ if (tableVersion > 8) {
+ assertEquals("bytes",
tableSchema.getField("dec_plain_large").schema().getType().getName());
+ assertEquals("decimal",
tableSchema.getField("dec_plain_large").schema().getLogicalType().getName());
+ assertEquals(20, ((LogicalTypes.Decimal)
tableSchema.getField("dec_plain_large").schema().getLogicalType()).getPrecision());
+ assertEquals(10, ((LogicalTypes.Decimal)
tableSchema.getField("dec_plain_large").schema().getLogicalType()).getScale());
+ }
+ assertEquals("fixed",
tableSchema.getField("dec_fixed_small").schema().getType().getName());
+ assertEquals(3,
tableSchema.getField("dec_fixed_small").schema().getFixedSize());
+ assertEquals("decimal",
tableSchema.getField("dec_fixed_small").schema().getLogicalType().getName());
+ assertEquals(5, ((LogicalTypes.Decimal)
tableSchema.getField("dec_fixed_small").schema().getLogicalType()).getPrecision());
+ assertEquals(2, ((LogicalTypes.Decimal)
tableSchema.getField("dec_fixed_small").schema().getLogicalType()).getScale());
+ assertEquals("fixed",
tableSchema.getField("dec_fixed_large").schema().getType().getName());
+ assertEquals(8,
tableSchema.getField("dec_fixed_large").schema().getFixedSize());
+ assertEquals("decimal",
tableSchema.getField("dec_fixed_large").schema().getLogicalType().getName());
+ assertEquals(18, ((LogicalTypes.Decimal)
tableSchema.getField("dec_fixed_large").schema().getLogicalType()).getPrecision());
+ assertEquals(9, ((LogicalTypes.Decimal)
tableSchema.getField("dec_fixed_large").schema().getLogicalType()).getScale());
+
+ sqlContext.clearCache();
+ Dataset<Row> df = sqlContext.read()
+ .options(hudiOpts)
+ .format("org.apache.hudi")
+ .load(tableBasePath);
+
+ long totalCount = df.count();
+ long expectedHalf = totalCount / 2;
+ long tolerance = totalCount / 20;
+ if (totalCount < 100) {
+ tolerance = totalCount / 4;
+ }
Review Comment:
Should we also validate the column stats index entries directly against
parquet min/max stats?
--
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]
