voonhous commented on code in PR #19384:
URL: https://github.com/apache/hudi/pull/19384#discussion_r3673426427
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java:
##########
@@ -170,9 +170,10 @@ private static boolean isTypeUpdateAllowInternal(Type src,
Type dst, boolean all
|| dst == Types.DoubleType.get() || dst == Types.StringType.get()
|| dst.typeId() == Type.TypeID.DECIMAL || dst.typeId() ==
Type.TypeID.DECIMAL_FIXED;
case LONG:
if (allowTimestampPrecisionEvolution
- && (dst.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MILLIS ||
dst.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MICROS)) {
- // Forward-fix path: 0.x stored local-timestamp columns as bare long
because its converter
- // did not recognize the logical type. Allow attaching the logical
type when the gate is open.
+ && (dst.typeId() == Type.TypeID.TIMESTAMP || dst.typeId() ==
Type.TypeID.TIMESTAMP_MILLIS
Review Comment:
The read side was never widened to match this. Both repair implementations
fix up "file column is a bare long, table says timestamp" **only when the table
type is local**:
- `SchemaRepair.java:151` -- `&& !((TimestampLogicalTypeAnnotation)
tableLogicalTypeAnnotation).isAdjustedToUTC()`
- `HoodieSchemaRepair.java:126` -- `return !tableTimestamp.isUtcAdjusted();`
Compiled `hudi-common` at this commit and ran the two
`needsLogicalTypeRepair` paths directly:
```
file bare long -> table timestamp-micros (UTC) repaired=false ->
"long"
file bare long -> table local-timestamp-micros repaired=true ->
local-timestamp-micros
file timestamp-micros -> table timestamp-millis repaired=true ->
timestamp-millis
```
That restriction came in with `fd79a1682e7e` (#14161) and was right at the
time: 0.x only dropped the *local* logical types, so "bare-long file +
UTC-timestamp table" could not arise. This PR makes it arise.
`TestHoodieSchemaRepair` has
`testRepairLongWithoutLogicalTypeToLocalTimestampMillis` / `...Micros` and no
UTC equivalent for the same reason.
Net effect: once a UTC promotion is authorized, every base file written
before it holds unannotated INT64 under a `timestamp-micros` table schema and
nothing fixes it up on read. It also makes the untouched line at
`HoodieCommonConfig.java:105` -- "Existing base files keep the old logical
type; Hudi readers compensate for it" -- false for exactly the case this PR
adds.
Two ways out, either works:
1. Drop `TIMESTAMP` / `TIMESTAMP_MILLIS` from this allowance and keep the
promotion local-only.
2. Widen `needsLogicalTypeRepair` in both classes to cover unannotated INT64
-> UTC timestamp, mirror
`testRepairLongWithoutLogicalTypeToLocalTimestampMillis` for the UTC targets,
and reword the `HoodieCommonConfig` line so it stops promising reader
compensation that is not there. Note
`HoodieFileGroupReaderBasedFileFormat.scala:95` gates repair on
`hasTimestampMillisField`, so that gate needs loosening too.
The read-back I asked for on the streamer test will tell you which of these
you are dealing with.
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java:
##########
@@ -94,9 +94,10 @@ public class HoodieCommonConfig extends HoodieConfig {
+ "entry is pinned to that logical type: an incoming value of a
different precision is coerced to it, and "
+ "the change from the table's current type is permitted. A
timestamp precision change with no entry for "
+ "the field is rejected with an error, so an unverified
micros/millis flip can never happen silently. "
- + "An entry also attaches a local-timestamp logical type to a column
that 0.x persisted as a bare long "
- + "because its converter did not recognize the type. A UTC/local
zone change is never authorized by "
- + "this config, whatever the entry says, since no rescale can
express it. "
+ + "An entry also attaches a timestamp logical type (UTC or local) to
a column persisted as a bare "
+ + "long, including one that 0.x stored without a logical type
because its converter did not recognize "
+ + "it. A UTC/local zone change is never authorized by this config,
whatever the entry says, since no "
Review Comment:
This promise is breakable after the change, on the paths that call
`reconcileSchema` without the Spark pre-pass.
`isTypeUpdateAllow(LONG, TIMESTAMP, true)` returns `true` now; on master it
returned `false`, and that `false` is what rejected the case below. Take a
bare-long table column, an incoming `local-timestamp-micros` field, and
`col:timestamp-micros` in the override:
1. `isGatedTimestampChange(LONG, local-timestamp-micros)` -> `true`
2. `applyTimestampOverrideOrThrow` is called with `skipIfEquals = tableType
= LONG`, the override is not `LONG`, so it calls
`typeChange.updateColumnType(col, TimestampType)` on a field that is still
`LONG`
3. `TableChanges.java:97` gates that on `isTypeUpdateAllow(LONG, TIMESTAMP,
true)` -- `true` after this PR, so it goes through
The table schema ends up UTC while the writer declares local.
`isCrossZoneTimestampChange` never fires because it needs timestamps on both
sides (`SchemaChangeUtils.java:119-121`) and the table side is a bare long.
Reachable from `BaseHoodieWriteClient.java:375`,
`HoodieMergeHelper.java:174` and `FileGroupReaderBasedMergeHandle.java:284` --
all three read the override from the write config, and none of them run
`reconcileTimestampLogicalType` first (that call exists only at
`HoodieSchemaUtils.scala:156`, the Spark datasource path).
Suggest rejecting in `applyTimestampOverrideOrThrow` when the override's
zone disagrees with the incoming field's zone, raising
`crossZoneTimestampChangeError` rather than the generic "incompatible type"
message, and adding a `reconcileSchema(table=long,
incoming=local-timestamp-micros, override=timestamp-micros)` case to
`TestAvroSchemaEvolutionUtils`.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -760,6 +767,97 @@ public void testTimestampMillis() throws Exception {
assertEquals(0,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
< '1980-01-01'").count());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testLongToTimestampPromotionGated(boolean setNullForMissingColumns)
throws Exception {
+ // Promoting a plain long column to a timestamp logical type is
override-gated: rejected without a
+ // per-field override for every target type, and applied with one. A bare
long carries no precision
+ // signal, so the override is the explicit verdict that authorizes the
promotion. One bare-long seed
+ // is reused: the rejection cases all throw (table stays bare long), and
the accepted case runs last.
+ String tableBasePath = basePath + "/testLongToTs" +
setNullForMissingColumns;
+ defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
+
+ // Sync 0: seed the table with `seconds_since_epoch` stored as a bare long.
+ HoodieDeltaStreamer.Config seed = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT,
+ Collections.singletonList(TestIdentityTransformer.class.getName()),
PROPS_FILENAME_TEST_SOURCE,
+ false, true, false, null, HoodieTableType.COPY_ON_WRITE.name());
+ seed.configs.add("hoodie.streamer.schemaprovider.source.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.streamer.schemaprovider.target.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.datasource.write.row.writer.enable=false");
+ seed.configs.add(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key() +
"=" + setNullForMissingColumns);
+ new HoodieDeltaStreamer(seed, jsc).sync();
+
+ Schema tableSchema = new
TableSchemaResolver(HoodieTestUtils.createMetaClient(storage, tableBasePath))
+ .getTableSchema(false).toAvroSchema();
+
assertNull(tableSchema.getField("seconds_since_epoch").schema().getLogicalType(),
+ "seconds_since_epoch must be seeded as a bare long in the table");
+ Schema baseSchema = new Schema.Parser().parse(fs.open(new Path(basePath +
"/source-timestamp-millis.avsc")));
+
+ // Every target type is rejected without a per-field override.
+ for (LogicalType targetType : new LogicalType[]
{LogicalTypes.timestampMillis(), LogicalTypes.timestampMicros(),
+ LogicalTypes.localTimestampMillis(),
LogicalTypes.localTimestampMicros()}) {
+ String schemaFile = writePromotedSchema(baseSchema, targetType,
setNullForMissingColumns);
+ HoodieDeltaStreamer.Config reject = promoteConfig(tableBasePath,
schemaFile, setNullForMissingColumns, null);
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(reject, jsc);
+ // sync() wraps the guard's SchemaCompatibilityException in a
HoodieIngestionException, so walk
+ // the cause chain to assert on the underlying exception.
+ Throwable thrown = assertThrows(Exception.class, streamer::sync,
+ "long -> " + targetType.getName() + " must be rejected without an
override");
+ Throwable cause = thrown;
+ while (cause != null && !(cause instanceof
SchemaCompatibilityException)) {
+ cause = cause.getCause();
+ }
+ assertTrue(cause instanceof SchemaCompatibilityException,
+ "Expected a SchemaCompatibilityException in the cause chain, got: "
+ thrown);
+ Type toType =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("field:" +
targetType.getName()).get("field");
+ assertEquals(AvroSchemaEvolutionUtils.timestampPrecisionChangeError(
+ "seconds_since_epoch", Types.LongType.get(), toType).getMessage(),
cause.getMessage());
+ }
+
+ // With an override the promotion is authorized (local promotions are
covered end-to-end by
+ // testCOWLogicalRepair / testMORLogicalRepair); verify a UTC promotion
succeeds and lands on the
+ // table schema.
+ String utcSchemaFile = writePromotedSchema(baseSchema,
LogicalTypes.timestampMicros(), setNullForMissingColumns);
+ HoodieDeltaStreamer.Config accept = promoteConfig(tableBasePath,
utcSchemaFile, setNullForMissingColumns,
+ "seconds_since_epoch:timestamp-micros");
+ new HoodieDeltaStreamer(accept, jsc).sync();
+ Schema evolved = new
TableSchemaResolver(HoodieTestUtils.createMetaClient(storage, tableBasePath))
+ .getTableSchema(false).toAvroSchema();
+ assertEquals("timestamp-micros",
evolved.getField("seconds_since_epoch").schema().getLogicalType().getName());
Review Comment:
This asserts the schema landed but never reads the table, so the state
described in my `SchemaChangeUtils` comment -- sync-0 base files holding
unannotated INT64 under a `timestamp-micros` table schema -- is created here
and then never put through a reader. The write goes down the Avro path, which
passes raw INT64 straight through, so the test stays green whether or not the
table is readable afterwards.
Every neighbour reads back: `testTimestampMillis:765` does
`sqlContext.read()...load(tableBasePath).filter("current_ts >
'1980-01-01'").count()`, and `testCOWLogicalRepair` from #19029 does
`assertDataframe(df, 15, 15)`.
Please add the read-back:
```suggestion
assertEquals("timestamp-micros",
evolved.getField("seconds_since_epoch").schema().getLogicalType().getName());
// The sync-0 base files still hold unannotated INT64 under a
timestamp-micros table schema --
// make sure they are still readable after the promotion.
sqlContext.clearCache();
sqlContext.read().format("org.apache.hudi").load(tableBasePath).select("seconds_since_epoch").collect();
```
One thing to be aware of if you also want to assert values:
`seconds_since_epoch` is seeded from `rand.nextLong()`
(`HoodieTestDataGenerator.java:582`), so any value assertion is meaningless
unless you seed it with a deterministic epoch-micros value first.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -760,6 +767,97 @@ public void testTimestampMillis() throws Exception {
assertEquals(0,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
< '1980-01-01'").count());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
Review Comment:
Two things about the cost of this test.
**The parameterization is inert for 5 of the 6 syncs it runs.**
`HoodieSchemaUtils.scala:156` calls `reconcileTimestampLogicalType`
unconditionally, *before* the `if (shouldReconcileSchema)` branch and before
`setNullForMissingColumns` is read at `:188` / `:226`. All four rejection cases
throw in that upfront guard, so the flag is never consulted; the seed sync hits
`latestTableSchemaOpt == None` at `:121`, also flag-independent. Only the
accept leg differs. So the boolean doubles a heavyweight Spark test to re-run
five identical syncs.
**There is a cheaper home that already exists.**
`TestHoodieSchemaUtils.java:343-359` (hudi-spark-common, no SparkContext)
already exposes `deduceWriterSchema(incoming, table, setNullForMissingColumns)`
backed by `TypedProperties`, and `:324-341` is already a `@ParameterizedTest
@ValueSource(booleans = {true, false})` over the same flag.
`TestHoodieDeltaStreamer` has 99 test methods and runs in the 120-minute "UT
Hudi Streamer & FT utilities" Azure job. Config threading and exception
surfacing end-to-end are already proven by the `testCOWLogicalRepair` /
`testMORLogicalRepair` rows.
Suggest moving the 4-target reject matrix into `TestHoodieSchemaUtils` with
the override added to the props, and keeping at most one unparameterized E2E
smoke case here -- seed, one UTC reject, one accept, with the read-back from my
other comment. That is 3 syncs instead of 12. Same complaint has been actioned
before in `960c3955892c` (#10492) and `eeccdf9bb0f2` (#10795).
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -760,6 +767,97 @@ public void testTimestampMillis() throws Exception {
assertEquals(0,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
< '1980-01-01'").count());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testLongToTimestampPromotionGated(boolean setNullForMissingColumns)
throws Exception {
+ // Promoting a plain long column to a timestamp logical type is
override-gated: rejected without a
+ // per-field override for every target type, and applied with one. A bare
long carries no precision
+ // signal, so the override is the explicit verdict that authorizes the
promotion. One bare-long seed
+ // is reused: the rejection cases all throw (table stays bare long), and
the accepted case runs last.
+ String tableBasePath = basePath + "/testLongToTs" +
setNullForMissingColumns;
+ defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
+
+ // Sync 0: seed the table with `seconds_since_epoch` stored as a bare long.
+ HoodieDeltaStreamer.Config seed = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT,
+ Collections.singletonList(TestIdentityTransformer.class.getName()),
PROPS_FILENAME_TEST_SOURCE,
+ false, true, false, null, HoodieTableType.COPY_ON_WRITE.name());
+ seed.configs.add("hoodie.streamer.schemaprovider.source.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.streamer.schemaprovider.target.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.datasource.write.row.writer.enable=false");
+ seed.configs.add(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key() +
"=" + setNullForMissingColumns);
+ new HoodieDeltaStreamer(seed, jsc).sync();
+
+ Schema tableSchema = new
TableSchemaResolver(HoodieTestUtils.createMetaClient(storage, tableBasePath))
+ .getTableSchema(false).toAvroSchema();
+
assertNull(tableSchema.getField("seconds_since_epoch").schema().getLogicalType(),
+ "seconds_since_epoch must be seeded as a bare long in the table");
+ Schema baseSchema = new Schema.Parser().parse(fs.open(new Path(basePath +
"/source-timestamp-millis.avsc")));
+
+ // Every target type is rejected without a per-field override.
+ for (LogicalType targetType : new LogicalType[]
{LogicalTypes.timestampMillis(), LogicalTypes.timestampMicros(),
+ LogicalTypes.localTimestampMillis(),
LogicalTypes.localTimestampMicros()}) {
+ String schemaFile = writePromotedSchema(baseSchema, targetType,
setNullForMissingColumns);
+ HoodieDeltaStreamer.Config reject = promoteConfig(tableBasePath,
schemaFile, setNullForMissingColumns, null);
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(reject, jsc);
+ // sync() wraps the guard's SchemaCompatibilityException in a
HoodieIngestionException, so walk
+ // the cause chain to assert on the underlying exception.
+ Throwable thrown = assertThrows(Exception.class, streamer::sync,
+ "long -> " + targetType.getName() + " must be rejected without an
override");
+ Throwable cause = thrown;
+ while (cause != null && !(cause instanceof
SchemaCompatibilityException)) {
Review Comment:
This cause-chain walk is now in the file three times -- here, `:1284` and
`:1435`, the latter two from #19029.
Nit: worth extracting one helper while you are in here, something like
```java
private static SchemaCompatibilityException
assertSchemaCompatCause(Executable executable) { ... }
```
and using it at all three sites.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java:
##########
@@ -277,7 +277,12 @@ private static SchemaCompatibilityException
crossZoneTimestampChangeError(String
col, from, to,
HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key()));
}
- private static SchemaCompatibilityException
timestampPrecisionChangeError(String col, Type from, Type to) {
+ /**
+ * Builds the actionable error for a gated timestamp logical-type change
with no per-field override
+ * in {@code hoodie.write.timestamp.logical.type.overrides}. Public so tests
can assert the exact
+ * message without duplicating its format.
+ */
+ public static SchemaCompatibilityException
timestampPrecisionChangeError(String col, Type from, Type to) {
Review Comment:
Worth tagging rather than explaining in prose -- the repo has
`org.apache.hudi.common.util.VisibleForTesting` and 22 files in
`hudi-common/src/main` use it, including `InternalSchemaConverter.java:31,112`
in this same package tree.
```suggestion
@VisibleForTesting
public static SchemaCompatibilityException
timestampPrecisionChangeError(String col, Type from, Type to) {
```
and drop the "Public so tests can assert the exact message without
duplicating its format." sentence from the javadoc, since the annotation says
it. `public` rather than package-private is right here, the consumer is in
`hudi-utilities`.
For what it is worth, I checked whether asserting the message by calling the
producer is tautological and it is not -- pre-PR the same call throws `Cannot
update column 'ts' from type 'long' to incompatible type 'timestamp'.` from
`TableChanges.java:98`, so the assertion does discriminate.
##########
hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java:
##########
@@ -840,4 +841,103 @@ public void testCrossZoneTimestampChangeIsRejected() {
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-millis")).toAvroSchema();
Assertions.assertEquals("local-timestamp-millis",
stillWorks.getField("ts").schema().getLogicalType().getName());
}
+
+ @Test
+ void testLongToUtcTimestampGatedInBothReconcilePaths() {
+ // Bare long to a UTC timestamp is override-gated exactly like the
local-timestamp case: rejected
+ // without a per-field override and applied with one, in both reconcile
paths. The non-reconcile
+ // guard previously skipped this and let it through silently on the
default write path.
+ HoodieSchema tableBareLong =
HoodieSchema.fromAvroSchema(tripAvro(Schema.create(Schema.Type.LONG)));
+ HoodieSchema incomingMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+
+ // No override: rejected in both paths with the exact actionable error.
+ Map<String, Type> noOverride =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("");
+ String expectedError =
AvroSchemaEvolutionUtils.timestampPrecisionChangeError(
+ "ts", Types.LongType.get(), Types.TimestampType.get()).getMessage();
+ SchemaCompatibilityException reconcileError =
assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incomingMicros,
tableBareLong, false, noOverride));
+ assertEquals(expectedError, reconcileError.getMessage());
+ SchemaCompatibilityException guardError =
assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingMicros,
tableBareLong, noOverride));
+ assertEquals(expectedError, guardError.getMessage());
+
+ // With the override: the promotion is applied in both paths.
+ Schema viaReconcile =
AvroSchemaEvolutionUtils.reconcileSchema(incomingMicros, tableBareLong, false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ assertEquals("timestamp-micros",
viaReconcile.getField("ts").schema().getLogicalType().getName());
+ Schema viaGuard =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingMicros,
tableBareLong,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ assertEquals("timestamp-micros",
viaGuard.getField("ts").schema().getLogicalType().getName());
+ }
+
+ @Test
+ void testLongToLocalTimestampGatedInBothReconcilePaths() {
Review Comment:
This one guards #19029's behaviour rather than this PR's. I spliced the two
new test files onto `d6ca44ed1279^` (plus only the `private -> public` line so
they compile) and ran them: `TestAvroSchemaEvolutionUtils` gives `Tests run:
22, Failures: 2` and the two failures are
`testLongToUtcTimestampGatedInBothReconcilePaths:859` and
`testNestedLongToTimestampGated:908`. **This test passes against pre-PR code.**
Its `reconcileSchema` legs also restate what
`testReconcileSchemaTimestampPrecisionEvolution` already asserts at `:693-704`
-- bare long rejected with no override, applied with
`ts:local-timestamp-millis` / `ts:local-timestamp-micros`.
The only genuinely uncovered bit is the `reconcileTimestampLogicalType` leg
for the local case. Please delete this test and add those two lines to the
existing block at `:693-704` instead.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -760,6 +767,97 @@ public void testTimestampMillis() throws Exception {
assertEquals(0,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
< '1980-01-01'").count());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testLongToTimestampPromotionGated(boolean setNullForMissingColumns)
throws Exception {
+ // Promoting a plain long column to a timestamp logical type is
override-gated: rejected without a
+ // per-field override for every target type, and applied with one. A bare
long carries no precision
+ // signal, so the override is the explicit verdict that authorizes the
promotion. One bare-long seed
+ // is reused: the rejection cases all throw (table stays bare long), and
the accepted case runs last.
+ String tableBasePath = basePath + "/testLongToTs" +
setNullForMissingColumns;
+ defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
+
+ // Sync 0: seed the table with `seconds_since_epoch` stored as a bare long.
+ HoodieDeltaStreamer.Config seed = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT,
+ Collections.singletonList(TestIdentityTransformer.class.getName()),
PROPS_FILENAME_TEST_SOURCE,
+ false, true, false, null, HoodieTableType.COPY_ON_WRITE.name());
+ seed.configs.add("hoodie.streamer.schemaprovider.source.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.streamer.schemaprovider.target.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.datasource.write.row.writer.enable=false");
Review Comment:
This line does nothing. The streamer's row-writer switch is
`hoodie.streamer.write.row.writer.enable` (`StreamSync.java:688`) and it only
applies to `BULK_INSERT`; this test uses INSERT and UPSERT. It looks like it
was copied from `testTimestampMillis:723`.
Harmless, but it reads as if the test is deliberately steering around the
row-writer path when it is not -- and the row-writer path does get the gate
anyway (`HoodieSparkSqlWriter.scala:466` computes the writer schema before the
`:515` short-circuit). Nit, feel free to ignore, but I would drop it from both
configs.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -760,6 +767,97 @@ public void testTimestampMillis() throws Exception {
assertEquals(0,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
< '1980-01-01'").count());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testLongToTimestampPromotionGated(boolean setNullForMissingColumns)
throws Exception {
+ // Promoting a plain long column to a timestamp logical type is
override-gated: rejected without a
+ // per-field override for every target type, and applied with one. A bare
long carries no precision
+ // signal, so the override is the explicit verdict that authorizes the
promotion. One bare-long seed
+ // is reused: the rejection cases all throw (table stays bare long), and
the accepted case runs last.
+ String tableBasePath = basePath + "/testLongToTs" +
setNullForMissingColumns;
+ defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
+
+ // Sync 0: seed the table with `seconds_since_epoch` stored as a bare long.
+ HoodieDeltaStreamer.Config seed = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT,
+ Collections.singletonList(TestIdentityTransformer.class.getName()),
PROPS_FILENAME_TEST_SOURCE,
+ false, true, false, null, HoodieTableType.COPY_ON_WRITE.name());
+ seed.configs.add("hoodie.streamer.schemaprovider.source.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.streamer.schemaprovider.target.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.datasource.write.row.writer.enable=false");
+ seed.configs.add(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key() +
"=" + setNullForMissingColumns);
+ new HoodieDeltaStreamer(seed, jsc).sync();
+
+ Schema tableSchema = new
TableSchemaResolver(HoodieTestUtils.createMetaClient(storage, tableBasePath))
+ .getTableSchema(false).toAvroSchema();
+
assertNull(tableSchema.getField("seconds_since_epoch").schema().getLogicalType(),
+ "seconds_since_epoch must be seeded as a bare long in the table");
+ Schema baseSchema = new Schema.Parser().parse(fs.open(new Path(basePath +
"/source-timestamp-millis.avsc")));
+
+ // Every target type is rejected without a per-field override.
+ for (LogicalType targetType : new LogicalType[]
{LogicalTypes.timestampMillis(), LogicalTypes.timestampMicros(),
+ LogicalTypes.localTimestampMillis(),
LogicalTypes.localTimestampMicros()}) {
+ String schemaFile = writePromotedSchema(baseSchema, targetType,
setNullForMissingColumns);
+ HoodieDeltaStreamer.Config reject = promoteConfig(tableBasePath,
schemaFile, setNullForMissingColumns, null);
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(reject, jsc);
+ // sync() wraps the guard's SchemaCompatibilityException in a
HoodieIngestionException, so walk
+ // the cause chain to assert on the underlying exception.
+ Throwable thrown = assertThrows(Exception.class, streamer::sync,
+ "long -> " + targetType.getName() + " must be rejected without an
override");
+ Throwable cause = thrown;
+ while (cause != null && !(cause instanceof
SchemaCompatibilityException)) {
+ cause = cause.getCause();
+ }
+ assertTrue(cause instanceof SchemaCompatibilityException,
+ "Expected a SchemaCompatibilityException in the cause chain, got: "
+ thrown);
+ Type toType =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("field:" +
targetType.getName()).get("field");
+ assertEquals(AvroSchemaEvolutionUtils.timestampPrecisionChangeError(
+ "seconds_since_epoch", Types.LongType.get(), toType).getMessage(),
cause.getMessage());
+ }
+
+ // With an override the promotion is authorized (local promotions are
covered end-to-end by
+ // testCOWLogicalRepair / testMORLogicalRepair); verify a UTC promotion
succeeds and lands on the
+ // table schema.
+ String utcSchemaFile = writePromotedSchema(baseSchema,
LogicalTypes.timestampMicros(), setNullForMissingColumns);
+ HoodieDeltaStreamer.Config accept = promoteConfig(tableBasePath,
utcSchemaFile, setNullForMissingColumns,
+ "seconds_since_epoch:timestamp-micros");
+ new HoodieDeltaStreamer(accept, jsc).sync();
+ Schema evolved = new
TableSchemaResolver(HoodieTestUtils.createMetaClient(storage, tableBasePath))
+ .getTableSchema(false).toAvroSchema();
+ assertEquals("timestamp-micros",
evolved.getField("seconds_since_epoch").schema().getLogicalType().getName());
+ }
+
+ private String writePromotedSchema(Schema baseSchema, LogicalType
targetType, boolean setNull) throws IOException {
+ Schema incoming = replaceFieldType(baseSchema, "seconds_since_epoch",
+ targetType.addToSchema(Schema.create(Schema.Type.LONG)));
+ String schemaFile = basePath + "/promote-" + targetType.getName() + "-nul"
+ setNull + ".avsc";
+ UtilitiesTestBase.Helpers.saveStringsToDFS(new String[]
{incoming.toString()}, storage, schemaFile);
+ return schemaFile;
+ }
+
+ private HoodieDeltaStreamer.Config promoteConfig(String tableBasePath,
String schemaFile,
+ boolean
setNullForMissingColumns, String override) {
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.UPSERT,
+ Collections.singletonList(TestIdentityTransformer.class.getName()),
PROPS_FILENAME_TEST_SOURCE,
+ false, true, false, null, HoodieTableType.COPY_ON_WRITE.name());
Review Comment:
Both this and the seed config hard-code `COPY_ON_WRITE`, so the MOR side of
the new promotion is untested. `testMORLogicalRepair` does cover MOR +
compaction + parquet/avro log blocks, but only for the precision flip and the
`long -> local` promotion (`LOGICAL_REPAIR_TS_OVERRIDES`).
The MOR branch matters here specifically: `HoodieAvroDataBlock.java:199`
calls the same `HoodieSchemaRepair.repairLogicalTypes(writerSchema,
readerSchema)` that is a no-op for UTC targets, so a pre-promotion log block
read under a post-promotion reader schema goes unrepaired.
Either parameterize this test over `{COPY_ON_WRITE, MERGE_ON_READ}`, or add
one MOR row to `testMORLogicalRepair` with a bare-long seed and
`seconds_since_epoch:timestamp-micros` and inline compaction on.
##########
hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java:
##########
@@ -840,4 +841,103 @@ public void testCrossZoneTimestampChangeIsRejected() {
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-millis")).toAvroSchema();
Assertions.assertEquals("local-timestamp-millis",
stillWorks.getField("ts").schema().getLogicalType().getName());
}
+
+ @Test
+ void testLongToUtcTimestampGatedInBothReconcilePaths() {
+ // Bare long to a UTC timestamp is override-gated exactly like the
local-timestamp case: rejected
+ // without a per-field override and applied with one, in both reconcile
paths. The non-reconcile
+ // guard previously skipped this and let it through silently on the
default write path.
+ HoodieSchema tableBareLong =
HoodieSchema.fromAvroSchema(tripAvro(Schema.create(Schema.Type.LONG)));
+ HoodieSchema incomingMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+
+ // No override: rejected in both paths with the exact actionable error.
+ Map<String, Type> noOverride =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("");
+ String expectedError =
AvroSchemaEvolutionUtils.timestampPrecisionChangeError(
+ "ts", Types.LongType.get(), Types.TimestampType.get()).getMessage();
+ SchemaCompatibilityException reconcileError =
assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incomingMicros,
tableBareLong, false, noOverride));
+ assertEquals(expectedError, reconcileError.getMessage());
+ SchemaCompatibilityException guardError =
assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingMicros,
tableBareLong, noOverride));
+ assertEquals(expectedError, guardError.getMessage());
+
+ // With the override: the promotion is applied in both paths.
+ Schema viaReconcile =
AvroSchemaEvolutionUtils.reconcileSchema(incomingMicros, tableBareLong, false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ assertEquals("timestamp-micros",
viaReconcile.getField("ts").schema().getLogicalType().getName());
+ Schema viaGuard =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingMicros,
tableBareLong,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ assertEquals("timestamp-micros",
viaGuard.getField("ts").schema().getLogicalType().getName());
+ }
+
+ @Test
+ void testLongToLocalTimestampGatedInBothReconcilePaths() {
+ // Bare long to local timestamp is override-gated (not forbidden):
rejected without an override
+ // and applied with one, and the non-reconcile guard must agree with
reconcileSchema.
+ HoodieSchema tableBareLong =
HoodieSchema.fromAvroSchema(tripAvro(Schema.create(Schema.Type.LONG)));
+ HoodieSchema incomingLocalMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+
+ // No override: rejected in both paths with the exact actionable error.
+ Map<String, Type> noOverride =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("");
+ String expectedError =
AvroSchemaEvolutionUtils.timestampPrecisionChangeError(
+ "ts", Types.LongType.get(),
Types.LocalTimestampMicrosType.get()).getMessage();
+ SchemaCompatibilityException reconcileError =
assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incomingLocalMicros,
tableBareLong, false, noOverride));
+ assertEquals(expectedError, reconcileError.getMessage());
+ SchemaCompatibilityException guardError =
assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingLocalMicros,
tableBareLong, noOverride));
+ assertEquals(expectedError, guardError.getMessage());
+ Schema repaired =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingLocalMicros,
tableBareLong,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-micros")).toAvroSchema();
+ assertEquals("local-timestamp-micros",
repaired.getField("ts").schema().getLogicalType().getName());
+ }
+
+ @Test
+ void testNestedLongToTimestampGated() {
+ // The gate resolves fully-qualified column names, so it applies to nested
fields too. A nested
+ // long -> timestamp (UTC or local) is override-gated via the dotted-key
override.
+ for (String token : new String[] {"timestamp-micros",
"local-timestamp-millis"}) {
Review Comment:
`local-timestamp-millis` here tests pre-existing behaviour -- narrow the
loop to just that token, run it against `d6ca44ed1279^`, and you get `Tests
run: 2, Failures: 0`. Only the `timestamp-micros` iteration is new.
Dropping it makes the loop single-token, which lets you delete `logicalLong`
entirely and inline the one type you need -- that also resolves the naming nit
the bot raised, better than renaming it. `HoodieSchema.createTimestampMicros()`
(`HoodieSchema.java:684`) is right there if you want a factory. Similarly
`nestedTrip` re-implements `HoodieSchemaTestUtils.createRecord` +
`createNestedField` from the same module.
Separately, worth spending the lines you save here on array and map
elements. I checked and the gate does reach them --
`InternalSchema.getAllColsFullName()` emits `arr.element` and `mp.value`, and
the dotted override applies on both paths:
```
cols=[arr, arr.element, mp, mp.value, mp.key, id]
guard threw: ... column 'arr.element' from 'long' to 'timestamp' ...
guard out arr =
{"type":"array","items":{"type":"long","logicalType":"timestamp-micros"}}
guard out mp =
{"type":"map","values":{"type":"long","logicalType":"timestamp-micros"}}
```
So the behaviour is fine, but nothing pins it and nothing tells a user that
an array element is addressed as `arr.element` and a map value as `mp.value`.
Please add those two cases here and one sentence naming the `element` / `value`
path segments in the config doc.
--
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]
