This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c1e2d0b0fcdf fix(schema): gate timestamp-precision change behind a
per-field verdict (#19029)
c1e2d0b0fcdf is described below
commit c1e2d0b0fcdf755547342e8856098080e7034e17
Author: Y Ethan Guo <[email protected]>
AuthorDate: Tue Jul 28 05:13:34 2026 -0700
fix(schema): gate timestamp-precision change behind a per-field verdict
(#19029)
* fix(schema): gate timestamp-precision change behind a per-field verdict
Earlier Hudi versions mishandled long-backed timestamp logical types in
AvroInternalSchemaConverter: timestamp-millis/-micros collapsed and were
re-emitted as
timestamp-micros, and the local-timestamp variants were dropped to a bare
long. Current
converters recognize all four as distinct, so the writer schema now
declares the correct
logical type; on every subsequent write the reconcile path finds the
mismatch and rejects it.
The correct target precision cannot be inferred from the schemas alone:
micros vs millis
depends on the stored longs, not on what the incoming schema declares. A
schema-only
decision could flip a genuinely-micros table to millis without rescaling
the values, after
which they read back as a wildly wrong instant (year ~58466). So the change
is authorized
per field with a verified target rather than toggled globally.
Changes:
- New advanced write config hoodie.write.timestamp.logical.type.overrides
(per-field target:
timestamp-{micros,millis} / local-timestamp-{micros,millis}). No entry
for a gated
precision change is rejected with an actionable
SchemaCompatibilityException; an entry
equal to the table type coerces the incoming values, a different entry
evolves the column.
- AvroSchemaEvolutionUtils.reconcileSchema takes the parsed override map;
new
reconcileTimestampLogicalType guards every writer-schema deduction path
(including the
default set.null=false path, whose Avro compatibility check is
logical-type-blind),
applied once in HoodieSchemaUtils.deduceWriterSchema.
- SchemaChangeUtils.parseTimestampLogicalTypeOverrides /
isGatedTimestampChange.
- New TimestampLogicalTypeClassifier (hudi-common): a pure value-based
verdict
(UNAFFECTED / CORRECT / LEGACY_0X_BUG / DROPPED_LOGICAL_TYPE / DIVERGENT
/ AMBIGUOUS) an
inspection tool can use to derive the correct override value from the
stored longs.
- Callers (BaseHoodieWriteClient, HoodieMergeHelper,
FileGroupReaderBasedMergeHandle) and
HoodieWriteConfig build-time validation read/parse the config.
Tests: TestTimestampLogicalTypeClassifier, TestSchemaChangeUtils,
TestAvroSchemaEvolutionUtils
(per-field reject/coerce/evolve plus a value-level assertion that a pinned
writer schema still
rescales the long by 1000), and TestHoodieDeltaStreamer COW/MOR
logical-repair (a verdict
authorizes the repair on the default and reconcile paths; no verdict is
rejected).
* docs(schema): make the timestamp-override guidance self-service and state
the file-rewrite step
Two P0 doc gaps on the timestamp logical type override config.
1. Both the config documentation and the SchemaCompatibilityException
pointed
the user at "the timestamp inspection tool", which does not exist in
Hudi.
The only such tool lives in a private repo, so a user who hit the
exception
had no way to comply. Replace it with the derivation rule itself: judge
the
stored longs, not the incoming schema, using the non-overlapping ~1e12 vs
~1e15 plausibility ranges that TimestampLogicalTypeClassifier already
implements.
2. Neither place said that the override corrects the table schema only.
Existing base files keep the old logical type and are not rewritten, so
external engines (Trino, Athena, BigQuery external, Spark-native parquet)
keep misreading them even though Hudi's own reader compensates. Document
the second migration step: rewrite the affected files via clustering or
compaction.
---------
Co-authored-by: voon <[email protected]>
---
.../apache/hudi/client/BaseHoodieWriteClient.java | 6 +-
.../org/apache/hudi/config/HoodieWriteConfig.java | 6 +
.../hudi/io/FileGroupReaderBasedMergeHandle.java | 5 +-
.../table/action/commit/HoodieMergeHelper.java | 5 +-
.../hudi/common/config/HoodieCommonConfig.java | 22 ++
.../schema/internal/action/TableChanges.java | 16 +-
.../internal/utils/AvroSchemaEvolutionUtils.java | 135 ++++++++++++-
.../schema/internal/utils/SchemaChangeUtils.java | 120 ++++++++++-
.../util/TimestampLogicalTypeClassifier.java | 218 ++++++++++++++++++++
.../utils/TestAvroSchemaEvolutionUtils.java | 223 ++++++++++++++++++++-
.../internal/utils/TestSchemaChangeUtils.java | 123 ++++++++++++
.../util/TestTimestampLogicalTypeClassifier.java | 143 +++++++++++++
.../scala/org/apache/hudi/HoodieSchemaUtils.scala | 35 +++-
.../deltastreamer/TestHoodieDeltaStreamer.java | 97 ++++++++-
14 files changed, 1114 insertions(+), 40 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index a3229c5bb72f..c43c9bcaa070 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -57,6 +57,7 @@ import
org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter;
import
org.apache.hudi.common.schema.internal.io.FileBasedInternalSchemaStorageManager;
import org.apache.hudi.common.schema.internal.utils.AvroSchemaEvolutionUtils;
import org.apache.hudi.common.schema.internal.utils.InternalSchemaUtils;
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils;
import org.apache.hudi.common.schema.internal.utils.SerDeHelper;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -371,7 +372,10 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
internalSchema =
InternalSchemaUtils.searchSchema(Long.parseLong(instantTime),
SerDeHelper.parseSchemas(historySchemaStr));
}
- InternalSchema evolvedSchema =
AvroSchemaEvolutionUtils.reconcileSchema(schema, internalSchema,
config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS));
+ InternalSchema evolvedSchema =
AvroSchemaEvolutionUtils.reconcileSchema(schema, internalSchema,
+
config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS),
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+
config.getStringOrDefault(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES)));
if (evolvedSchema.equals(internalSchema)) {
metadata.addMetadata(SerDeHelper.LATEST_SCHEMA,
SerDeHelper.toJson(evolvedSchema));
//TODO save history schema by metaTable
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index 7aadac3d3407..c145192262e2 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -52,6 +52,7 @@ import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.WriteConcurrencyMode;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
@@ -3874,6 +3875,11 @@ public class HoodieWriteConfig extends HoodieConfig {
+ "schedule inline compaction (%s) can be enabled. Both can't be
set to true at the same time. %s, %s",
HoodieCompactionConfig.INLINE_COMPACT.key(),
HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT.key(), inlineCompact,
inlineCompactSchedule));
+ // Parse-and-discard so a malformed 'field:type' entry fails at client
build time rather
+ // than deep inside deduceWriterSchema on the first commit. Empty
(default) is a no-op.
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+
writeConfig.getStringOrDefault(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES));
+
int lookbackCommits =
writeConfig.getInt(ROLLING_METADATA_TIMELINE_LOOKBACK_COMMITS);
checkArgument(lookbackCommits >= 0,
String.format("%s must be non-negative, but was %d",
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
index beba9067c7e6..460ef5d2b4fc 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
@@ -41,6 +41,7 @@ import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.internal.InternalSchema;
import org.apache.hudi.common.schema.internal.utils.AvroSchemaEvolutionUtils;
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils;
import org.apache.hudi.common.schema.internal.utils.SerDeHelper;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.cdc.HoodieCDCUtils;
@@ -281,7 +282,9 @@ public class FileGroupReaderBasedMergeHandle<T, I, K, O>
extends HoodieWriteMerg
boolean usePosition =
config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
Option<InternalSchema> internalSchemaOption =
SerDeHelper.fromJson(config.getInternalSchema())
.map(internalSchema ->
AvroSchemaEvolutionUtils.reconcileSchema(writeSchemaWithMetaFields,
internalSchema,
-
config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS)));
+
config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS),
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+
config.getStringOrDefault(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES))));
long maxMemoryPerCompaction = getMaxMemoryForMerge();
props.put(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE.key(),
String.valueOf(maxMemoryPerCompaction));
Option<Stream<HoodieLogFile>> logFilesStreamOpt =
compactionOperation.map(op -> op.getDeltaFileNames().stream().map(logFileName ->
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
index 96a6cedb3812..d04cf143640c 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
@@ -28,6 +28,7 @@ import
org.apache.hudi.common.schema.internal.action.InternalSchemaMerger;
import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter;
import org.apache.hudi.common.schema.internal.utils.AvroSchemaEvolutionUtils;
import org.apache.hudi.common.schema.internal.utils.InternalSchemaUtils;
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils;
import org.apache.hudi.common.schema.internal.utils.SerDeHelper;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.TableSchemaResolver;
@@ -171,7 +172,9 @@ public class HoodieMergeHelper<T> extends BaseMergeHelper {
if (querySchemaOpt.isPresent() &&
!baseFile.getBootstrapBaseFile().isPresent()) {
// check implicitly add columns, and position reorder(spark sql may
change cols order)
InternalSchema querySchema =
AvroSchemaEvolutionUtils.reconcileSchema(writerSchema,
- querySchemaOpt.get(),
writeConfig.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS));
+ querySchemaOpt.get(),
writeConfig.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS),
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+
writeConfig.getStringOrDefault(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES)));
long commitInstantTime = Long.parseLong(baseFile.getCommitTime());
InternalSchema fileSchema =
InternalSchemaCache.getInternalSchemaByVersionId(commitInstantTime, metaClient);
if (fileSchema.isEmptySchema() &&
writeConfig.getBoolean(HoodieCommonConfig.RECONCILE_SCHEMA)) {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
index f41423d6e412..6724128d3e7f 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
@@ -83,6 +83,28 @@ public class HoodieCommonConfig extends HoodieConfig {
+ " operation will fail schema compatibility check. Set this option
to true will make the missing "
+ " column be filled with null values to successfully complete the
write operation.");
+ public static final ConfigProperty<String> TIMESTAMP_LOGICAL_TYPE_OVERRIDES
= ConfigProperty
+ .key("hoodie.write.timestamp.logical.type.overrides")
+ .defaultValue("")
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Per-field authority for the timestamp logical type,
taking precedence over the "
+ + "auto-inferred schema. Comma-separated 'field:type' pairs, where
type is one of timestamp-micros, "
+ + "timestamp-millis, local-timestamp-micros, local-timestamp-millis
(case-insensitive). A field with an "
+ + "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. "
+ + "Derive the value from the stored longs, never from the incoming
schema: for instants after 1990 an "
+ + "epoch-millis value is around 1e12 while epoch-micros is around
1e15, so the two ranges do not "
+ + "overlap. TimestampLogicalTypeClassifier implements that verdict
for inspection tooling to reuse. "
+ + "NOTE: this corrects the table schema only. Existing base files
keep the old logical type; Hudi "
+ + "readers compensate for it, but external engines (Trino, Athena,
BigQuery external, Spark-native "
+ + "parquet) keep misreading those files until they are rewritten
under the corrected schema via "
+ + "clustering or compaction. Treat this as a one-time migration: set
the override, then rewrite.");
+
public static final ConfigProperty<ExternalSpillableMap.DiskMapType>
SPILLABLE_DISK_MAP_TYPE = ConfigProperty
.key("hoodie.common.spillable.diskmap.type")
.defaultValue(ExternalSpillableMap.DiskMapType.BITCASK)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/action/TableChanges.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/action/TableChanges.java
index b54ee3782eeb..29db547dc01d 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/action/TableChanges.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/action/TableChanges.java
@@ -47,13 +47,11 @@ public class TableChanges {
@Getter
private final Map<Integer, Types.Field> updates = new HashMap<>();
+ private final boolean allowTimestampPrecisionEvolution;
- private ColumnUpdateChange(InternalSchema schema) {
- super(schema, false);
- }
-
- private ColumnUpdateChange(InternalSchema schema, boolean caseSensitive) {
+ private ColumnUpdateChange(InternalSchema schema, boolean caseSensitive,
boolean allowTimestampPrecisionEvolution) {
super(schema, caseSensitive);
+ this.allowTimestampPrecisionEvolution = allowTimestampPrecisionEvolution;
}
@Override
@@ -96,7 +94,7 @@ public class TableChanges {
throw new SchemaCompatibilityException(String.format("Cannot update
type for column '%s' because it does not exist in the schema", name));
}
- if (!SchemaChangeUtils.isTypeUpdateAllow(field.type(), newType)) {
+ if (!SchemaChangeUtils.isTypeUpdateAllow(field.type(), newType,
allowTimestampPrecisionEvolution)) {
throw new SchemaCompatibilityException(String.format(
"Cannot update column '%s' from type '%s' to incompatible type
'%s'.", name, field.type(), newType));
}
@@ -232,11 +230,11 @@ public class TableChanges {
}
public static ColumnUpdateChange get(InternalSchema schema) {
- return new ColumnUpdateChange(schema);
+ return new ColumnUpdateChange(schema, false, false);
}
- public static ColumnUpdateChange get(InternalSchema schema, boolean
caseSensitive) {
- return new ColumnUpdateChange(schema, caseSensitive);
+ public static ColumnUpdateChange get(InternalSchema schema, boolean
caseSensitive, boolean allowTimestampPrecisionEvolution) {
+ return new ColumnUpdateChange(schema, caseSensitive,
allowTimestampPrecisionEvolution);
}
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
index f6405fef0137..947fc72df797 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
@@ -18,17 +18,22 @@
package org.apache.hudi.common.schema.internal.utils;
+import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.schema.internal.InternalSchema;
+import org.apache.hudi.common.schema.internal.Type;
import org.apache.hudi.common.schema.internal.action.TableChanges;
import org.apache.hudi.common.schema.internal.action.TableChangesHelper;
+import org.apache.hudi.exception.SchemaCompatibilityException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.stream.Collectors;
@@ -62,7 +67,8 @@ public class AvroSchemaEvolutionUtils {
* nullable in the result. Otherwise, no
updates will be made to those fields.
* @return reconcile Schema
*/
- public static InternalSchema reconcileSchema(HoodieSchema incomingSchema,
InternalSchema oldTableSchema, boolean makeMissingFieldsNullable) {
+ public static InternalSchema reconcileSchema(HoodieSchema incomingSchema,
InternalSchema oldTableSchema,
+ boolean
makeMissingFieldsNullable, Map<String, Type> timestampLogicalTypeOverrides) {
/* If incoming schema is null, we fall back on table schema. */
if (incomingSchema.isSchemaNull()) {
return oldTableSchema;
@@ -129,9 +135,22 @@ public class AvroSchemaEvolutionUtils {
// do type evolution.
InternalSchema internalSchemaAfterAddColumns =
SchemaChangeUtils.applyTableChanges2Schema(oldTableSchema, addChange);
- TableChanges.ColumnUpdateChange typeChange =
TableChanges.ColumnUpdateChange.get(internalSchemaAfterAddColumns);
+ // The reconcile pre-validates timestamp precision changes per field below
(against the explicit
+ // overrides), so the update change is constructed permissively;
non-overridden precision changes
+ // are rejected here with an actionable error rather than deferred to the
gate.
+ TableChanges.ColumnUpdateChange typeChange =
TableChanges.ColumnUpdateChange.get(
+ internalSchemaAfterAddColumns, false, true);
typeChangeColumns.stream().filter(f ->
!inComingInternalSchema.findType(f).isNestedType()).forEach(col -> {
- typeChange.updateColumnType(col, inComingInternalSchema.findType(col));
+ Type tableType = oldTableSchema.findType(col);
+ Type incomingType = inComingInternalSchema.findType(col);
+ if (SchemaChangeUtils.isGatedTimestampChange(tableType, incomingType)) {
+ // Skip-if equals the *table* type: the reconcile is producing the new
table schema starting
+ // from oldTableSchema, so a coerce-to-table-precision override needs
no schema update — the
+ // writer coerces incoming values via rewriteRecordWithNewSchema.
+ applyTimestampOverrideOrThrow(col, tableType, incomingType,
timestampLogicalTypeOverrides, tableType, typeChange);
+ } else {
+ typeChange.updateColumnType(col, incomingType);
+ }
});
// relax existing columns to nullable when the incoming schema made them
nullable (valid widening)
@@ -162,8 +181,114 @@ public class AvroSchemaEvolutionUtils {
return evolvedSchema;
}
- public static HoodieSchema reconcileSchema(HoodieSchema incomingSchema,
HoodieSchema oldTableSchema, boolean makeMissingFieldsNullable) {
- return convert(reconcileSchema(incomingSchema, convert(oldTableSchema),
makeMissingFieldsNullable), oldTableSchema.getFullName());
+ public static HoodieSchema reconcileSchema(HoodieSchema incomingSchema,
HoodieSchema oldTableSchema, boolean makeMissingFieldsNullable,
+ Map<String, Type>
timestampLogicalTypeOverrides) {
+ return convert(reconcileSchema(incomingSchema, convert(oldTableSchema),
makeMissingFieldsNullable, timestampLogicalTypeOverrides),
oldTableSchema.getFullName());
+ }
+
+ /**
+ * Reconciles only the timestamp logical-type precision of {@code
writerSchema} against
+ * {@code tableSchema}, independent of column add/drop/nullability
reconciliation. This is the
+ * single guard that every writer-schema deduction path must apply,
including the non-reconcile
+ * paths that otherwise validate via the logical-type-blind Avro
reader/writer compatibility check
+ * and would let an unverified micros/millis flip through silently.
+ *
+ * <p>For each field whose precision differs from the table: an override
pins it (equal to the
+ * table type coerces the incoming values, a different type applies the
authorized evolution), and
+ * a change with no override throws. A UTC/local zone change throws
unconditionally, since no
+ * override authorizes one. Non-timestamp changes are left untouched here.
+ */
+ public static HoodieSchema reconcileTimestampLogicalType(HoodieSchema
writerSchema, HoodieSchema tableSchema,
+ Map<String, Type>
timestampLogicalTypeOverrides) {
+ if (writerSchema == null || writerSchema.getType() !=
HoodieSchemaType.RECORD
+ || tableSchema == null || tableSchema.getType() !=
HoodieSchemaType.RECORD) {
+ return writerSchema;
+ }
+ InternalSchema writerInternal = convert(writerSchema);
+ InternalSchema tableInternal = convert(tableSchema);
+ List<String> tableCols = tableInternal.getAllColsFullName();
+ TableChanges.ColumnUpdateChange typeChange =
TableChanges.ColumnUpdateChange.get(writerInternal, false, true);
+ boolean changed = false;
+ for (String col : writerInternal.getAllColsFullName()) {
+ if (!tableCols.contains(col)) {
+ continue;
+ }
+ Type writerType = writerInternal.findType(col);
+ Type tableType = tableInternal.findType(col);
+ if (writerType.isNestedType()) {
+ continue;
+ }
+ // A zone change is never authorizable, and this is the only guard on
the default
+ // non-reconcile path -- the Avro reader/writer check that follows is
logical-type-blind for
+ // two long-backed fields, so skipping here would let the flip through
silently.
+ if (SchemaChangeUtils.isCrossZoneTimestampChange(tableType, writerType))
{
+ throw crossZoneTimestampChangeError(col, tableType, writerType);
+ }
+ if (!SchemaChangeUtils.isGatedTimestampChange(tableType, writerType)) {
+ continue;
+ }
+ // Skip-if equals the *writer* type: this method returns a modified
writerSchema. When the
+ // override already matches the writer field, the writer schema is what
we want; no update.
+ if (applyTimestampOverrideOrThrow(col, tableType, writerType,
timestampLogicalTypeOverrides, writerType, typeChange)) {
+ changed = true;
+ }
+ }
+ if (!changed) {
+ return writerSchema;
+ }
+ return convert(SchemaChangeUtils.applyTableChanges2Schema(writerInternal,
typeChange), writerSchema.getFullName());
+ }
+
+ /**
+ * Shared override-apply for a single field whose type is a gated timestamp
precision change.
+ * Called from both {@link #reconcileSchema} and {@link
#reconcileTimestampLogicalType} — those
+ * two paths differ only in which "current" schema they compare the override
against (the table
+ * type vs. the writer type), so the caller passes that in as {@code
skipIfEquals}.
+ *
+ * @param col the fully-qualified column name (for the error
message)
+ * @param tableType the table's current type (for the error message)
+ * @param incomingType the writer/incoming type (for the error message)
+ * @param overrides the parsed per-field overrides map
+ * @param skipIfEquals compare the override against this; no schema
update when equal
+ * @param typeChange the accumulator for schema updates
+ * @return {@code true} if the override was applied (schema will change),
{@code false} otherwise
+ * @throws SchemaCompatibilityException when no override is present for this
gated change
+ */
+ private static boolean applyTimestampOverrideOrThrow(String col, Type
tableType, Type incomingType,
+ Map<String, Type>
overrides, Type skipIfEquals,
+
TableChanges.ColumnUpdateChange typeChange) {
+ Type overrideType = overrides.get(col);
+ if (overrideType == null) {
+ throw timestampPrecisionChangeError(col, tableType, incomingType);
+ }
+ if (overrideType.equals(skipIfEquals)) {
+ return false;
+ }
+ typeChange.updateColumnType(col, overrideType);
+ return true;
+ }
+
+ private static SchemaCompatibilityException
crossZoneTimestampChangeError(String col, Type from, Type to) {
+ return new SchemaCompatibilityException(String.format(
+ "Refusing to change the timestamp logical type of column '%s' from
'%s' to '%s': this crosses the "
+ + "UTC/local boundary, which changes the instant the stored value
denotes and cannot be repaired by "
+ + "rescaling. '%s' authorizes precision changes only, never a zone
change. Keep writing the column "
+ + "with its existing zone, or add a new column and backfill it
with an explicit conversion.",
+ col, from, to,
HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key()));
+ }
+
+ private static SchemaCompatibilityException
timestampPrecisionChangeError(String col, Type from, Type to) {
+ return new SchemaCompatibilityException(String.format(
+ "Refusing to change the timestamp logical type of column '%s' from
'%s' to '%s' without an explicit "
+ + "verdict. This precision change is not applied automatically
because the correct target depends "
+ + "on the stored values, not the incoming schema. Inspect the raw
long values of '%s' in the existing "
+ + "base files: for instants after 1990 epoch-millis is around 1e12
and epoch-micros is around 1e15, "
+ + "so the ranges do not overlap (TimestampLogicalTypeClassifier
implements this verdict). Then set "
+ + "'%s' to the precision the values actually are, for example
'%s:timestamp-micros' to keep the "
+ + "current precision and coerce the incoming values, or
'%s:timestamp-millis' to evolve the column. "
+ + "Existing base files are not rewritten by this change; rewrite
them via clustering or compaction "
+ + "so that non-Hudi readers also see the corrected type.",
+ col, from, to, col,
HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key(), col, col));
}
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java
index b2eabe5b035c..567c1397f431 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java
@@ -28,7 +28,11 @@ import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
+import java.util.Map;
/**
* Helper methods for schema Change.
@@ -36,6 +40,96 @@ import java.util.List;
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class SchemaChangeUtils {
+ /**
+ * Parses the {@code hoodie.write.timestamp.logical.type.overrides} value
into a per-field map of
+ * the target timestamp {@link Type}. The value is a comma-separated list of
{@code field:type}
+ * pairs, where type is one of timestamp-micros, timestamp-millis,
local-timestamp-micros,
+ * local-timestamp-millis (case-insensitive). The tokens are a Hudi-owned
vocabulary decoupled
+ * from any serialization format.
+ *
+ * <p>Splits on the last {@code ':'} so dotted nested field names ({@code
parent.child}) work
+ * unchanged. Field names containing a literal {@code ':'} are not supported.
+ *
+ * @param value the raw config value (may be null or empty)
+ * @return an unmodifiable map from field name to the pinned timestamp type;
empty if unset
+ */
+ public static Map<String, Type> parseTimestampLogicalTypeOverrides(String
value) {
+ if (value == null || value.trim().isEmpty()) {
+ return Collections.emptyMap();
+ }
+ Map<String, Type> result = new LinkedHashMap<>();
+ for (String pair : value.split(",")) {
+ String trimmed = pair.trim();
+ if (trimmed.isEmpty()) {
+ continue;
+ }
+ int sep = trimmed.lastIndexOf(':');
+ if (sep <= 0 || sep == trimmed.length() - 1) {
+ throw new IllegalArgumentException("Invalid timestamp logical type
override entry '" + trimmed
+ + "'. Expected 'field:type' where type is one of timestamp-micros,
timestamp-millis, "
+ + "local-timestamp-micros, local-timestamp-millis.");
+ }
+ String field = trimmed.substring(0, sep).trim();
+ Type type = timestampTypeFromToken(trimmed.substring(sep + 1).trim());
+ result.put(field, type);
+ }
+ return Collections.unmodifiableMap(result);
+ }
+
+ private static Type timestampTypeFromToken(String token) {
+ switch (token.toLowerCase(Locale.ROOT)) {
+ case "timestamp-micros":
+ return Types.TimestampType.get();
+ case "timestamp-millis":
+ return Types.TimestampMillisType.get();
+ case "local-timestamp-micros":
+ return Types.LocalTimestampMicrosType.get();
+ case "local-timestamp-millis":
+ return Types.LocalTimestampMillisType.get();
+ default:
+ throw new IllegalArgumentException("Unknown timestamp logical type
token '" + token
+ + "'. Expected one of timestamp-micros, timestamp-millis,
local-timestamp-micros, "
+ + "local-timestamp-millis.");
+ }
+ }
+
+ /**
+ * Whether a column type change is a timestamp precision change that must be
authorized by an
+ * explicit per-field override (see {@code
hoodie.write.timestamp.logical.type.overrides}). This
+ * covers timestamp-micros/millis flips, local-timestamp-micros/millis
flips, and the forward-fix
+ * from a bare {@code long} to a local-timestamp logical type that 0.x
dropped.
+ */
+ public static boolean isGatedTimestampChange(Type src, Type dst) {
+ if (src.equals(dst)) {
+ return false;
+ }
+ if (isUtcTimestamp(src) && isUtcTimestamp(dst)) {
+ return true;
+ }
+ if (isLocalTimestamp(src) && isLocalTimestamp(dst)) {
+ return true;
+ }
+ return src.typeId() == Type.TypeID.LONG && isLocalTimestamp(dst);
+ }
+
+ /**
+ * Whether a column type change crosses the UTC/local timestamp boundary.
Unlike a precision
+ * change this has no value-level repair: the same long denotes a different
instant under each
+ * interpretation, so rescaling cannot express the conversion. A zone change
is therefore always
+ * rejected and no per-field override authorizes it.
+ */
+ public static boolean isCrossZoneTimestampChange(Type src, Type dst) {
+ return (isUtcTimestamp(src) && isLocalTimestamp(dst)) ||
(isLocalTimestamp(src) && isUtcTimestamp(dst));
+ }
+
+ private static boolean isUtcTimestamp(Type type) {
+ return type.typeId() == Type.TypeID.TIMESTAMP || type.typeId() ==
Type.TypeID.TIMESTAMP_MILLIS;
+ }
+
+ private static boolean isLocalTimestamp(Type type) {
+ return type.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MILLIS ||
type.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MICROS;
+ }
+
/**
* Whether to allow the column type to be updated.
* now only support:
@@ -52,29 +146,35 @@ public class SchemaChangeUtils {
* @param dst new column type.
* @return whether to allow the column type to be updated.
*/
- public static boolean isTypeUpdateAllow(Type src, Type dst) {
+ public static boolean isTypeUpdateAllow(Type src, Type dst, boolean
allowTimestampPrecisionEvolution) {
if (src.isNestedType() || dst.isNestedType()) {
throw new IllegalArgumentException("only support update primitive type");
}
if (src.equals(dst)) {
return true;
}
- return isTypeUpdateAllowInternal(src, dst);
+ return isTypeUpdateAllowInternal(src, dst,
allowTimestampPrecisionEvolution);
}
public static boolean shouldPromoteType(Type src, Type dst) {
if (src.equals(dst) || src.isNestedType() || dst.isNestedType()) {
return false;
}
- return isTypeUpdateAllowInternal(src, dst);
+ return isTypeUpdateAllowInternal(src, dst, false);
}
- private static boolean isTypeUpdateAllowInternal(Type src, Type dst) {
+ private static boolean isTypeUpdateAllowInternal(Type src, Type dst, boolean
allowTimestampPrecisionEvolution) {
switch (src.typeId()) {
case INT:
return dst == Types.LongType.get() || dst == Types.FloatType.get()
|| 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.
+ return true;
+ }
return dst == Types.FloatType.get() || dst == Types.DoubleType.get()
|| dst == Types.StringType.get() || dst.typeId() == Type.TypeID.DECIMAL ||
dst.typeId() == Type.TypeID.DECIMAL_FIXED;
case FLOAT:
return dst == Types.DoubleType.get() || dst == Types.StringType.get()
|| dst.typeId() == Type.TypeID.DECIMAL || dst.typeId() ==
Type.TypeID.DECIMAL_FIXED;
@@ -90,6 +190,18 @@ public class SchemaChangeUtils {
return isDecimalFixedUpdateAllowInternal(src, dst);
case STRING:
return dst == Types.DateType.get() || dst.typeId() ==
Type.TypeID.DECIMAL || dst.typeId() == Type.TypeID.DECIMAL_FIXED || dst ==
Types.BinaryType.get();
+ case TIMESTAMP:
+ case TIMESTAMP_MILLIS:
+ if (!allowTimestampPrecisionEvolution) {
+ return false;
+ }
+ return dst.typeId() == Type.TypeID.TIMESTAMP || dst.typeId() ==
Type.TypeID.TIMESTAMP_MILLIS;
+ case LOCAL_TIMESTAMP_MILLIS:
+ case LOCAL_TIMESTAMP_MICROS:
+ if (!allowTimestampPrecisionEvolution) {
+ return false;
+ }
+ return dst.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MILLIS ||
dst.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MICROS;
default:
return false;
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/TimestampLogicalTypeClassifier.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/TimestampLogicalTypeClassifier.java
new file mode 100644
index 000000000000..efb2b6e5be2c
--- /dev/null
+++
b/hudi-common/src/main/java/org/apache/hudi/common/util/TimestampLogicalTypeClassifier.java
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.util;
+
+import org.apache.avro.LogicalType;
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+
+/**
+ * Shared classifier for the timestamp logical-type drift introduced by Hudi
0.14.1 / 0.15.0 / 1.0.x.
+ * It decides, from three signals about a long-backed column, what the correct
target timestamp
+ * logical type is, so a caller can suggest a value for
+ * {@code hoodie.write.timestamp.logical.type.overrides}.
+ *
+ * <p>The three signals are: (1) the table schema logical type, (2) the
base-file schema logical
+ * type, and (3) the shape of the raw stored {@code long} values. Signal (1)
must be resolved as of
+ * the file's own commit instant, not the latest table schema, so that a file
is classified within
+ * its own era. The classification here is pure; the I/O that gathers the
signals is caller-specific
+ * (the OSS scanner reads through the storage abstraction, the data-plane tool
reads parquet
+ * directly), but both must share this logic so their verdicts cannot drift
apart.
+ *
+ * <p><b>Status:</b> this is the verdict logic only. The inspection scanner
that samples base files
+ * and emits a ready-to-paste config value is not part of this repo yet, so an
in-repo grep finds
+ * only tests today. It lives here rather than in the tool so that every
consumer shares one
+ * definition of the verdict and they cannot drift apart.
+ *
+ * <p>Do not rely on the enums or method signatures here as stable public API:
this is internal to
+ * the timestamp-repair workflow.
+ */
+public class TimestampLogicalTypeClassifier {
+
+ private TimestampLogicalTypeClassifier() {
+ }
+
+ // Plausibility windows: an epoch instant in 1990-01-01 .. 2100-01-01,
interpreted as millis vs
+ // micros. The two windows are ~1000x apart and do not overlap, so a single
value fits at most one.
+ private static final long PLAUSIBLE_MILLIS_MIN = 631152000000L; //
1990-01-01
+ private static final long PLAUSIBLE_MILLIS_MAX = 4102444800000L; //
2100-01-01
+ private static final long PLAUSIBLE_MICROS_MIN = 631152000000000L; //
1990-01-01
+ private static final long PLAUSIBLE_MICROS_MAX = 4102444800000000L; //
2100-01-01
+
+ /** The timestamp logical type of a long-backed column, or NONE / UNKNOWN. */
+ public enum LogicalTimestampType {
+ NONE,
+ TIMESTAMP_MICROS,
+ TIMESTAMP_MILLIS,
+ LOCAL_TIMESTAMP_MICROS,
+ LOCAL_TIMESTAMP_MILLIS,
+ UNKNOWN
+ }
+
+ /** The shape of a stored long value, judged against the plausibility
windows. */
+ public enum DataShape {
+ MICROS,
+ MILLIS,
+ AMBIGUOUS,
+ UNKNOWN
+ }
+
+ /** The per-column verdict. */
+ public enum Bucket {
+ /** No timestamp logical type and no timestamp-shaped data. Nothing to do,
safe to upgrade. */
+ UNAFFECTED,
+ /** Table, file, and values agree at the same precision. Correct, though
it may still need a
+ * defensive pin if the ingestion source declares a different precision. */
+ CORRECT,
+ /**
+ * Label says micros but the values are millis, or the symmetric inverse
(label says millis but
+ * values are micros): the 0.14.1 drift. Both directions map to the same
repair action — pin the
+ * field to whatever the values actually are — so they share this bucket.
The observed
+ * production case is label_micros/values_millis; the symmetric case is
included to be safe.
+ */
+ LEGACY_0X_BUG,
+ /** Bare long, but the values are timestamp-shaped: the 0.x
local-timestamp logical-type loss. */
+ DROPPED_LOGICAL_TYPE,
+ /** The three signals disagree in some other way. */
+ DIVERGENT,
+ /** The value shape cannot be judged confidently (near-epoch, sentinels,
zeros, negatives). */
+ AMBIGUOUS
+ }
+
+ /** Classifies the logical type of a long-backed Avro field (the union is
expected to be unwrapped). */
+ public static LogicalTimestampType classifyAvroLogicalType(Schema
longSchema) {
+ LogicalType lt = longSchema.getLogicalType();
+ if (lt == null) {
+ return LogicalTimestampType.NONE;
+ }
+ if (lt instanceof LogicalTypes.TimestampMillis) {
+ return LogicalTimestampType.TIMESTAMP_MILLIS;
+ }
+ if (lt instanceof LogicalTypes.TimestampMicros) {
+ return LogicalTimestampType.TIMESTAMP_MICROS;
+ }
+ if (lt instanceof LogicalTypes.LocalTimestampMillis) {
+ return LogicalTimestampType.LOCAL_TIMESTAMP_MILLIS;
+ }
+ if (lt instanceof LogicalTypes.LocalTimestampMicros) {
+ return LogicalTimestampType.LOCAL_TIMESTAMP_MICROS;
+ }
+ return LogicalTimestampType.UNKNOWN;
+ }
+
+ /**
+ * Judges a single raw long. Zeros, negatives, and sentinels (for example
the year-9999 markers)
+ * fall outside both plausibility windows and are reported UNKNOWN so they
never drive a verdict.
+ */
+ public static DataShape classifyValueShape(long value) {
+ if (value <= 0) {
+ return DataShape.UNKNOWN;
+ }
+ boolean millisPlausible = value >= PLAUSIBLE_MILLIS_MIN && value <
PLAUSIBLE_MILLIS_MAX;
+ boolean microsPlausible = value >= PLAUSIBLE_MICROS_MIN && value <
PLAUSIBLE_MICROS_MAX;
+ if (millisPlausible && !microsPlausible) {
+ return DataShape.MILLIS;
+ }
+ if (microsPlausible && !millisPlausible) {
+ return DataShape.MICROS;
+ }
+ if (millisPlausible) {
+ // The windows do not overlap, so this is unreachable for a single
value; kept for safety.
+ return DataShape.AMBIGUOUS;
+ }
+ return DataShape.UNKNOWN;
+ }
+
+ /** Folds one sampled value's shape into the running per-column shape across
many samples/files. */
+ public static DataShape reduceShape(DataShape acc, DataShape sample) {
+ if (acc == null || acc == DataShape.UNKNOWN) {
+ return sample;
+ }
+ if (sample == DataShape.UNKNOWN || acc == sample) {
+ return acc;
+ }
+ return DataShape.AMBIGUOUS;
+ }
+
+ /**
+ * Reconciles the three signals into a verdict. {@code tableType} must be
the table logical type as
+ * of the inspected file's commit instant.
+ */
+ public static Bucket classifyBucket(LogicalTimestampType tableType,
LogicalTimestampType fileType, DataShape shape) {
+ boolean noLogicalType = tableType == LogicalTimestampType.NONE && fileType
== LogicalTimestampType.NONE;
+ if (shape == DataShape.UNKNOWN) {
+ // Nothing timestamp-shaped was seen. Bare-long-everywhere is a plain
non-timestamp column;
+ // anything else cannot be judged without a clearer value signal.
+ return noLogicalType ? Bucket.UNAFFECTED : Bucket.AMBIGUOUS;
+ }
+ if (shape == DataShape.AMBIGUOUS) {
+ return Bucket.AMBIGUOUS;
+ }
+ if (noLogicalType) {
+ // Bare long with timestamp-shaped data: 0.x dropped the local-timestamp
logical type.
+ return Bucket.DROPPED_LOGICAL_TYPE;
+ }
+ boolean tableMicros = tableType == LogicalTimestampType.TIMESTAMP_MICROS
|| tableType == LogicalTimestampType.LOCAL_TIMESTAMP_MICROS;
+ boolean tableMillis = tableType == LogicalTimestampType.TIMESTAMP_MILLIS
|| tableType == LogicalTimestampType.LOCAL_TIMESTAMP_MILLIS;
+ boolean fileMicros = fileType == LogicalTimestampType.TIMESTAMP_MICROS ||
fileType == LogicalTimestampType.LOCAL_TIMESTAMP_MICROS;
+ boolean fileMillis = fileType == LogicalTimestampType.TIMESTAMP_MILLIS ||
fileType == LogicalTimestampType.LOCAL_TIMESTAMP_MILLIS;
+ if (tableMicros && fileMicros && shape == DataShape.MILLIS) {
+ return Bucket.LEGACY_0X_BUG;
+ }
+ if (tableMillis && fileMillis && shape == DataShape.MICROS) {
+ return Bucket.LEGACY_0X_BUG;
+ }
+ if (tableMicros && fileMicros && shape == DataShape.MICROS) {
+ // All three agree on micros. Genuinely correct; a source-declared
millis is indistinguishable
+ // here and maps to the same action (keep micros), so it is not a
separate bucket.
+ return Bucket.CORRECT;
+ }
+ if (tableMillis && fileMillis && shape == DataShape.MILLIS) {
+ return Bucket.CORRECT;
+ }
+ // Reached when the table and file logical types disagree (for example
table_micros +
+ // file_millis) or the surviving cases where the three signals do not line
up to CORRECT
+ // or LEGACY_0X_BUG. The operator must decide the correct override; no
auto-suggestion.
+ return Bucket.DIVERGENT;
+ }
+
+ /**
+ * The suggested {@code hoodie.write.timestamp.logical.type.overrides} token
for a column, or empty
+ * when the operator must decide (ambiguous / divergent) or nothing is
needed (unaffected).
+ * {@code local} selects the local-timestamp variant, carried from the
table/file logical type.
+ */
+ public static Option<String> suggestedOverrideToken(Bucket bucket, DataShape
shape, boolean local) {
+ switch (bucket) {
+ case CORRECT:
+ // Pin to the current precision so a differently-declared source
cannot flip it.
+ return Option.of(token(shape, local));
+ case LEGACY_0X_BUG:
+ case DROPPED_LOGICAL_TYPE:
+ // Repair to what the values actually are.
+ return Option.of(token(shape, local));
+ default:
+ return Option.empty();
+ }
+ }
+
+ private static String token(DataShape shape, boolean local) {
+ String precision = shape == DataShape.MILLIS ? "millis" : "micros";
+ return (local ? "local-timestamp-" : "timestamp-") + precision;
+ }
+}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
index d82dd84b87e1..83f0e47af648 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
@@ -19,6 +19,7 @@
package org.apache.hudi.common.schema.internal.utils;
import org.apache.hudi.common.avro.HoodieAvroUtils;
+import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.schema.HoodieJsonProperties;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
@@ -31,6 +32,7 @@ import
org.apache.hudi.common.schema.internal.action.TableChanges;
import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter;
import org.apache.hudi.common.testutils.SchemaTestUtil;
import org.apache.hudi.exception.HoodieNullSchemaTypeException;
+import org.apache.hudi.exception.SchemaCompatibilityException;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
@@ -486,7 +488,8 @@ public class TestAvroSchemaEvolutionUtils {
);
evolvedRecord =
(Types.RecordType)InternalSchemaBuilder.getBuilder().refreshNewId(evolvedRecord,
new AtomicInteger(0));
HoodieSchema evolvedSchema =
InternalSchemaConverter.convert(evolvedRecord, "test1");
- InternalSchema result =
AvroSchemaEvolutionUtils.reconcileSchema(evolvedSchema, oldSchema, false);
+ InternalSchema result =
AvroSchemaEvolutionUtils.reconcileSchema(evolvedSchema, oldSchema, false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(""));
Types.RecordType checkedRecord = Types.RecordType.get(
Types.Field.get(0, false, "id", Types.IntType.get()),
Types.Field.get(1, true, "data", Types.StringType.get()),
@@ -541,7 +544,8 @@ public class TestAvroSchemaEvolutionUtils {
+
"{\"name\":\"d2\",\"type\":[\"null\",{\"type\":\"int\",\"logicalType\":\"date\"}],\"default\":null}]}");
HoodieSchema simpleReconcileSchema =
InternalSchemaConverter.convert(AvroSchemaEvolutionUtils
- .reconcileSchema(incomingSchema,
InternalSchemaConverter.convert(schema), false), "schemaNameFallback");
+ .reconcileSchema(incomingSchema,
InternalSchemaConverter.convert(schema), false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")),
"schemaNameFallback");
Assertions.assertEquals(simpleCheckSchema, simpleReconcileSchema);
}
@@ -563,7 +567,8 @@ public class TestAvroSchemaEvolutionUtils {
InternalSchema oldInternalSchema =
InternalSchemaConverter.convert(oldSchema);
// set a non-default schema id for old table schema, e.g., 2.
oldInternalSchema.setSchemaId(2);
- InternalSchema evolvedSchema =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldInternalSchema,
false);
+ InternalSchema evolvedSchema =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldInternalSchema,
false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(""));
// the evolved schema should be the old table schema, since there is no
type change at all.
Assertions.assertEquals(oldInternalSchema, evolvedSchema);
}
@@ -590,7 +595,8 @@ public class TestAvroSchemaEvolutionUtils {
incomingRecord = (Types.RecordType)
InternalSchemaBuilder.getBuilder().refreshNewId(incomingRecord, new
AtomicInteger(0));
HoodieSchema incomingSchema =
InternalSchemaConverter.convert(incomingRecord, "test1");
- InternalSchema result =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldSchema, true);
+ InternalSchema result =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldSchema, true,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(""));
Types.RecordType checkedRecord = Types.RecordType.get(
Types.Field.get(0, false, "id", Types.IntType.get()),
@@ -619,7 +625,8 @@ public class TestAvroSchemaEvolutionUtils {
incomingRecord = (Types.RecordType)
InternalSchemaBuilder.getBuilder().refreshNewId(incomingRecord, new
AtomicInteger(0));
HoodieSchema incomingSchema =
InternalSchemaConverter.convert(incomingRecord, "test1");
- InternalSchema result =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldSchema, true);
+ InternalSchema result =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldSchema, true,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(""));
Types.RecordType checkedRecord = Types.RecordType.get(
Types.Field.get(0, false, "id", Types.IntType.get()),
@@ -627,4 +634,210 @@ public class TestAvroSchemaEvolutionUtils {
);
Assertions.assertEquals(checkedRecord, result.getRecord());
}
+
+ private static Schema tripAvro(Schema tsType) {
+ return Schema.createRecord("trip", null, null, false, Arrays.asList(
+ new Schema.Field("id", Schema.create(Schema.Type.STRING), null, null),
+ new Schema.Field("ts", tsType, null, null)));
+ }
+
+ @Test
+ public void testReconcileSchemaTimestampPrecisionEvolution() {
+ // A timestamp precision change is rejected unless the field has an
explicit override in
+ // hoodie.write.timestamp.logical.type.overrides. The override pins the
field: an entry equal to
+ // the table type coerces the incoming values and keeps the table
precision, while a different
+ // entry evolves the column. No entry throws, so an unverified
micros/millis flip cannot happen.
+ HoodieSchema tableSchemaMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+ HoodieSchema incomingSchemaMillis =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG))));
+
+ // Guard: with no override, the precision change is rejected in either
direction with an
+ // actionable error that names the column and the config to set.
+ Throwable rejectedMicrosToMillis =
assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incomingSchemaMillis,
tableSchemaMicros, false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+ assertTrue(rejectedMicrosToMillis.getMessage().contains("without an
explicit"));
+
assertTrue(rejectedMicrosToMillis.getMessage().contains(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key()));
+ assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(tableSchemaMicros,
incomingSchemaMillis, false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+
+ // Override to millis: the micros table evolves to millis (the
genuine-repair case).
+ Schema evolvedToMillis =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchemaMillis,
tableSchemaMicros, false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-millis")).toAvroSchema();
+ Assertions.assertEquals("timestamp-millis",
evolvedToMillis.getField("ts").schema().getLogicalType().getName());
+
+ // Override to micros with a millis source (the Apna case): the table
stays micros, no flip; the
+ // incoming millis values are coerced to micros on write.
+ Schema pinnedToMicros =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchemaMillis,
tableSchemaMicros, false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ Assertions.assertEquals("timestamp-micros",
pinnedToMicros.getField("ts").schema().getLogicalType().getName());
+
+ // Override to micros against a millis table: the reverse evolution is
permitted.
+ Schema evolvedToMicros =
AvroSchemaEvolutionUtils.reconcileSchema(tableSchemaMicros,
incomingSchemaMillis, false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ Assertions.assertEquals("timestamp-micros",
evolvedToMicros.getField("ts").schema().getLogicalType().getName());
+
+ // The same override applies to the local-timestamp variants.
+ HoodieSchema tableLocalMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+ HoodieSchema incomingLocalMillis =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.localTimestampMillis().addToSchema(Schema.create(Schema.Type.LONG))));
+ assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incomingLocalMillis,
tableLocalMicros, false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+ Schema reconciledLocal =
AvroSchemaEvolutionUtils.reconcileSchema(incomingLocalMillis, tableLocalMicros,
false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-millis")).toAvroSchema();
+ Assertions.assertEquals("local-timestamp-millis",
reconciledLocal.getField("ts").schema().getLogicalType().getName());
+
+ // 0.x did not recognize the local-timestamp logical types, so affected
tables persisted those
+ // columns as bare long. The override must also allow attaching the
logical type on forward-fix.
+ HoodieSchema tableBareLong =
HoodieSchema.fromAvroSchema(tripAvro(Schema.create(Schema.Type.LONG)));
+ assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incomingLocalMillis,
tableBareLong, false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+ Schema repairedToLocalMillis =
AvroSchemaEvolutionUtils.reconcileSchema(incomingLocalMillis, tableBareLong,
false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-millis")).toAvroSchema();
+ Assertions.assertEquals("local-timestamp-millis",
repairedToLocalMillis.getField("ts").schema().getLogicalType().getName());
+
+ HoodieSchema incomingLocalMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+ Schema repairedToLocalMicros =
AvroSchemaEvolutionUtils.reconcileSchema(incomingLocalMicros, tableBareLong,
false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-micros")).toAvroSchema();
+ Assertions.assertEquals("local-timestamp-micros",
repairedToLocalMicros.getField("ts").schema().getLogicalType().getName());
+ }
+
+ @Test
+ public void testReconcileTimestampLogicalTypeGuardsNonReconcilePath() {
+ // reconcileTimestampLogicalType is the guard applied to the deduced
writer schema on every path,
+ // including the default set.null=false path whose Avro compatibility
check is logical-type-blind.
+ HoodieSchema tableMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+ HoodieSchema writerMillis =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG))));
+
+ // Guard: no override and the precision differs, so the flip is rejected
instead of silently applied.
+ Throwable rejected = assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(writerMillis,
tableMicros,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+ assertTrue(rejected.getMessage().contains("without an explicit"));
+ assertTrue(rejected.getMessage().contains("'ts'"));
+
+ // Override to micros coerces the millis writer back to micros (no flip,
the Apna case).
+ Schema coerced =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(writerMillis,
tableMicros,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ Assertions.assertEquals("timestamp-micros",
coerced.getField("ts").schema().getLogicalType().getName());
+
+ // Override to millis keeps the writer at millis (authorized evolution).
+ Schema evolved =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(writerMillis,
tableMicros,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-millis")).toAvroSchema();
+ Assertions.assertEquals("timestamp-millis",
evolved.getField("ts").schema().getLogicalType().getName());
+
+ // No precision difference: returned unchanged, no override required and
no throw.
+ Schema unchanged =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(tableMicros, tableMicros,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")).toAvroSchema();
+ Assertions.assertEquals("timestamp-micros",
unchanged.getField("ts").schema().getLogicalType().getName());
+ }
+
+ /**
+ * End-to-end value assertion on the coerce/pin path — the Apna case. Source
declares
+ * timestamp-millis, table is timestamp-micros, override pins the field to
the table's micros
+ * type. The reconcile flips the writer schema back to micros. When a record
whose source Avro
+ * schema declared millis is rewritten to the (now-micros) writer schema,
the long must still be
+ * rescaled by 1000 — not left as-is because writer == table.
+ *
+ * <p>The prior boolean flag would have flipped the table to millis without
touching values,
+ * causing the "reads as year 58466" failure. This test guards that
value-level behavior directly.
+ */
+ @Test
+ public void testReconcileTimestampLogicalTypeCoercesValuesOnPin() {
+ HoodieSchema tableMicrosSchema =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+ Schema sourceMillisSchema =
tripAvro(LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG)));
+
+ // Driver-plan step: apply the guard with the coerce override. The writer
schema for `ts`
+ // should be pinned back to timestamp-micros (matching the table), not
left as millis.
+ Schema writerSchema =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(
+ HoodieSchema.fromAvroSchema(sourceMillisSchema), tableMicrosSchema,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ Assertions.assertEquals("timestamp-micros",
writerSchema.getField("ts").schema().getLogicalType().getName());
+
+ // Executor step: an incoming record carrying the SOURCE schema (millis)
is rewritten to the
+ // deduced WRITER schema (micros). rewriteRecordWithNewSchema must invoke
the x1000 rescale so
+ // 2024-01-01T00:00:00Z millis (1704067200000L) becomes the equivalent
micros
+ // (1704067200000000L) — not the same long reinterpreted, which would read
as year 55965.
+ long millisValue = 1704067200000L; // 2024-01-01T00:00:00Z as epoch millis
+ long expectedMicros = 1704067200000000L; // same instant as epoch micros
+ GenericRecord sourceRecord = new GenericData.Record(sourceMillisSchema);
+ sourceRecord.put("id", "row-1");
+ sourceRecord.put("ts", millisValue);
+ GenericRecord rewritten =
HoodieAvroUtils.rewriteRecordWithNewSchema(sourceRecord, writerSchema);
+ Assertions.assertEquals(expectedMicros, rewritten.get("ts"),
+ "millis source value must be rescaled to micros when the writer schema
is pinned to micros");
+
+ // Symmetric coverage: source declares micros, table is millis, override
pins to millis.
+ // Rewrite must divide by 1000 (integer division). Pick a value that is
exact.
+ HoodieSchema tableMillisSchema =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG))));
+ Schema sourceMicrosSchema =
tripAvro(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG)));
+ Schema writerSchemaMillis =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(
+ HoodieSchema.fromAvroSchema(sourceMicrosSchema), tableMillisSchema,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-millis")).toAvroSchema();
+ Assertions.assertEquals("timestamp-millis",
writerSchemaMillis.getField("ts").schema().getLogicalType().getName());
+ GenericRecord sourceMicros = new GenericData.Record(sourceMicrosSchema);
+ sourceMicros.put("id", "row-2");
+ sourceMicros.put("ts", expectedMicros);
+ GenericRecord rewrittenMillis =
HoodieAvroUtils.rewriteRecordWithNewSchema(sourceMicros, writerSchemaMillis);
+ Assertions.assertEquals(millisValue, rewrittenMillis.get("ts"),
+ "micros source value must be rescaled to millis when the writer schema
is pinned to millis");
+ }
+
+ /**
+ * A UTC/local zone change is not a precision repair. The stored long means
a different instant
+ * under each interpretation and no rescale can fix that, so a zone change
must be rejected on
+ * every path and no per-field override may authorize it.
+ *
+ * <p>Both entry points have to enforce it. reconcileSchema rejects via
isTypeUpdateAllow, but
+ * reconcileTimestampLogicalType is the only guard on the default
non-reconcile path, and the
+ * Avro reader/writer compatibility check that runs after it is
logical-type-blind for two
+ * long-backed fields -- so if the guard skips a zone change, nothing else
catches it.
+ */
+ @Test
+ public void testCrossZoneTimestampChangeIsRejected() {
+ HoodieSchema tableMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+ HoodieSchema localMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+ HoodieSchema tableMillis =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG))));
+ HoodieSchema localMillis =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.localTimestampMillis().addToSchema(Schema.create(Schema.Type.LONG))));
+
+ // No override: rejected by both entry points, in both zone directions.
+ assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(localMicros,
InternalSchemaConverter.convert(tableMicros), false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+ assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(tableMicros,
InternalSchemaConverter.convert(localMicros), false,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+ Throwable guarded = assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(localMicros, tableMicros,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+ assertTrue(guarded.getMessage().contains("'ts'"), "Unexpected message: " +
guarded.getMessage());
+ assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(tableMicros, localMicros,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+
+ // An override must NOT unlock a zone change, whichever zone it names.
+ assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(localMicros, tableMicros,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-micros")));
+ assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(localMicros, tableMicros,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")));
+ assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(localMicros,
InternalSchemaConverter.convert(tableMicros), false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-micros")));
+
+ // A zone change that also crosses precision is still a zone change.
+ assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(localMillis, tableMicros,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-millis")));
+ assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(localMicros, tableMillis,
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("")));
+
+ // Same-zone precision changes are unaffected: still gated by the
override, not by the zone check.
+ Schema stillWorks =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(localMillis, localMicros,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-millis")).toAvroSchema();
+ Assertions.assertEquals("local-timestamp-millis",
stillWorks.getField("ts").schema().getLogicalType().getName());
+ }
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestSchemaChangeUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestSchemaChangeUtils.java
new file mode 100644
index 000000000000..9e0579214bea
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestSchemaChangeUtils.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.schema.internal.utils;
+
+import org.apache.hudi.common.schema.internal.Type;
+import org.apache.hudi.common.schema.internal.Types;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link SchemaChangeUtils#parseTimestampLogicalTypeOverrides(String)}.
Validation must
+ * happen at parse time — the config is threaded through the writer schema
deduction path and a
+ * malformed value would otherwise surface deep inside deduceWriterSchema on
the first commit.
+ */
+public class TestSchemaChangeUtils {
+
+ @Test
+ public void parseEmptyValueYieldsEmptyMap() {
+
assertTrue(SchemaChangeUtils.parseTimestampLogicalTypeOverrides(null).isEmpty());
+
assertTrue(SchemaChangeUtils.parseTimestampLogicalTypeOverrides("").isEmpty());
+ assertTrue(SchemaChangeUtils.parseTimestampLogicalTypeOverrides("
").isEmpty());
+ }
+
+ @Test
+ public void parseValidSingleEntry() {
+ Map<String, Type> overrides =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-millis");
+ assertEquals(1, overrides.size());
+ assertEquals(Types.TimestampMillisType.get(), overrides.get("ts"));
+ }
+
+ @Test
+ public void parseAllFourTokens() {
+ Map<String, Type> overrides =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+
"a:timestamp-micros,b:timestamp-millis,c:local-timestamp-micros,d:local-timestamp-millis");
+ assertEquals(4, overrides.size());
+ assertEquals(Types.TimestampType.get(), overrides.get("a"));
+ assertEquals(Types.TimestampMillisType.get(), overrides.get("b"));
+ assertEquals(Types.LocalTimestampMicrosType.get(), overrides.get("c"));
+ assertEquals(Types.LocalTimestampMillisType.get(), overrides.get("d"));
+ }
+
+ @Test
+ public void parseIsCaseInsensitive() {
+ // Tokens are case-insensitive so an operator pasting "Timestamp-Millis"
is not surprised.
+ assertEquals(Types.TimestampMillisType.get(),
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:TIMESTAMP-MILLIS").get("ts"));
+ assertEquals(Types.LocalTimestampMicrosType.get(),
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:Local-Timestamp-Micros").get("ts"));
+ }
+
+ @Test
+ public void parseSupportsDottedNestedFieldNames() {
+ // Nested field names use '.'; the parser splits on the LAST ':' so this
works unchanged.
+ Map<String, Type> overrides =
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("payload.event_time:timestamp-millis");
+ assertEquals(1, overrides.size());
+ assertEquals(Types.TimestampMillisType.get(),
overrides.get("payload.event_time"));
+ }
+
+ @Test
+ public void parseTrimmedWhitespaceAndSkipEmptySegments() {
+ Map<String, Type> overrides =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+ " a:timestamp-micros ,, b : timestamp-millis ,");
+ assertEquals(2, overrides.size());
+ assertEquals(Types.TimestampType.get(), overrides.get("a"));
+ assertEquals(Types.TimestampMillisType.get(), overrides.get("b"));
+ }
+
+ @Test
+ public void parseRejectsMissingColon() {
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () ->
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("field_only"));
+ assertTrue(ex.getMessage().contains("field_only"), "message should include
the offending entry");
+ }
+
+ @Test
+ public void parseRejectsMissingType() {
+ assertThrows(IllegalArgumentException.class,
+ () -> SchemaChangeUtils.parseTimestampLogicalTypeOverrides("field:"));
+ }
+
+ @Test
+ public void parseRejectsMissingField() {
+ assertThrows(IllegalArgumentException.class,
+ () ->
SchemaChangeUtils.parseTimestampLogicalTypeOverrides(":timestamp-micros"));
+ }
+
+ @Test
+ public void parseRejectsUnknownToken() {
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () ->
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("field:not-a-real-type"));
+ assertTrue(ex.getMessage().contains("not-a-real-type"), "message should
include the bad token");
+ }
+
+ @Test
+ public void parseResultIsUnmodifiable() {
+ Map<String, Type> overrides =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros");
+ assertThrows(UnsupportedOperationException.class,
+ () -> overrides.put("other", Types.TimestampMillisType.get()));
+ }
+}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/util/TestTimestampLogicalTypeClassifier.java
b/hudi-common/src/test/java/org/apache/hudi/common/util/TestTimestampLogicalTypeClassifier.java
new file mode 100644
index 000000000000..1a81fa2cb338
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/util/TestTimestampLogicalTypeClassifier.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.util;
+
+import org.apache.hudi.common.util.TimestampLogicalTypeClassifier.Bucket;
+import org.apache.hudi.common.util.TimestampLogicalTypeClassifier.DataShape;
+
+import org.junit.jupiter.api.Test;
+
+import static
org.apache.hudi.common.util.TimestampLogicalTypeClassifier.LogicalTimestampType.LOCAL_TIMESTAMP_MICROS;
+import static
org.apache.hudi.common.util.TimestampLogicalTypeClassifier.LogicalTimestampType.LOCAL_TIMESTAMP_MILLIS;
+import static
org.apache.hudi.common.util.TimestampLogicalTypeClassifier.LogicalTimestampType.NONE;
+import static
org.apache.hudi.common.util.TimestampLogicalTypeClassifier.LogicalTimestampType.TIMESTAMP_MICROS;
+import static
org.apache.hudi.common.util.TimestampLogicalTypeClassifier.LogicalTimestampType.TIMESTAMP_MILLIS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * Tests {@link TimestampLogicalTypeClassifier}.
+ */
+public class TestTimestampLogicalTypeClassifier {
+
+ // 2025-06-01 as millis (~13 digits) and micros (~16 digits).
+ private static final long MILLIS_2025 = 1748736000000L;
+ private static final long MICROS_2025 = 1748736000000000L;
+ // The year-9999 micros sentinel seen on real tables.
+ private static final long YEAR_9999_MICROS = 253402214400000000L;
+
+ @Test
+ public void testValueShape() {
+ assertEquals(DataShape.MILLIS,
TimestampLogicalTypeClassifier.classifyValueShape(MILLIS_2025));
+ assertEquals(DataShape.MICROS,
TimestampLogicalTypeClassifier.classifyValueShape(MICROS_2025));
+ // Zero, negative, near-epoch, and sentinels are not judgeable.
+ assertEquals(DataShape.UNKNOWN,
TimestampLogicalTypeClassifier.classifyValueShape(0L));
+ assertEquals(DataShape.UNKNOWN,
TimestampLogicalTypeClassifier.classifyValueShape(-1L));
+ assertEquals(DataShape.UNKNOWN,
TimestampLogicalTypeClassifier.classifyValueShape(1000L));
+ assertEquals(DataShape.UNKNOWN,
TimestampLogicalTypeClassifier.classifyValueShape(YEAR_9999_MICROS));
+ }
+
+ @Test
+ public void testReduceShape() {
+ assertEquals(DataShape.MICROS,
TimestampLogicalTypeClassifier.reduceShape(null, DataShape.MICROS));
+ assertEquals(DataShape.MICROS,
TimestampLogicalTypeClassifier.reduceShape(DataShape.MICROS, DataShape.MICROS));
+ // UNKNOWN samples do not pollute a settled shape.
+ assertEquals(DataShape.MICROS,
TimestampLogicalTypeClassifier.reduceShape(DataShape.MICROS,
DataShape.UNKNOWN));
+ assertEquals(DataShape.MICROS,
TimestampLogicalTypeClassifier.reduceShape(DataShape.UNKNOWN,
DataShape.MICROS));
+ // Genuinely mixed precision across files (for example a wrongly flipped
table) surfaces as ambiguous.
+ assertEquals(DataShape.AMBIGUOUS,
TimestampLogicalTypeClassifier.reduceShape(DataShape.MICROS, DataShape.MILLIS));
+ }
+
+ @Test
+ public void testBuckets() {
+ // Genuinely correct micros (the Apna case): all three signals agree.
+ assertEquals(Bucket.CORRECT,
TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MICROS,
TIMESTAMP_MICROS, DataShape.MICROS));
+ // Legit millis.
+ assertEquals(Bucket.CORRECT,
TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MILLIS,
TIMESTAMP_MILLIS, DataShape.MILLIS));
+ // The 0.14.1 drift: label micros, values millis.
+ assertEquals(Bucket.LEGACY_0X_BUG,
TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MICROS,
TIMESTAMP_MICROS, DataShape.MILLIS));
+ // Symmetric inverse.
+ assertEquals(Bucket.LEGACY_0X_BUG,
TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MILLIS,
TIMESTAMP_MILLIS, DataShape.MICROS));
+ // Dropped logical type: bare long, timestamp-shaped values.
+ assertEquals(Bucket.DROPPED_LOGICAL_TYPE,
TimestampLogicalTypeClassifier.classifyBucket(NONE, NONE, DataShape.MICROS));
+ // No logical type and no timestamp-shaped data: not a timestamp column at
all.
+ assertEquals(Bucket.UNAFFECTED,
TimestampLogicalTypeClassifier.classifyBucket(NONE, NONE, DataShape.UNKNOWN));
+ // A labeled timestamp column with an unjudgeable value shape cannot be
classified.
+ assertEquals(Bucket.AMBIGUOUS,
TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MICROS,
TIMESTAMP_MICROS, DataShape.UNKNOWN));
+ assertEquals(Bucket.AMBIGUOUS,
TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MICROS,
TIMESTAMP_MICROS, DataShape.AMBIGUOUS));
+ // Table and file disagree with no clean repair reading.
+ assertEquals(Bucket.DIVERGENT,
TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MICROS,
TIMESTAMP_MILLIS, DataShape.MICROS));
+ }
+
+ @Test
+ public void testSuggestedOverrideToken() {
+ assertEquals("timestamp-millis",
+
TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.LEGACY_0X_BUG,
DataShape.MILLIS, false).get());
+ assertEquals("timestamp-micros",
+ TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.CORRECT,
DataShape.MICROS, false).get());
+ assertEquals("local-timestamp-micros",
+
TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.DROPPED_LOGICAL_TYPE,
DataShape.MICROS, true).get());
+ assertEquals("local-timestamp-millis",
+
TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.DROPPED_LOGICAL_TYPE,
DataShape.MILLIS, true).get());
+ // Millis-side symmetry: DROPPED with millis data pins to timestamp-millis
/ local-timestamp-millis.
+ assertEquals("timestamp-millis",
+
TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.DROPPED_LOGICAL_TYPE,
DataShape.MILLIS, false).get());
+ assertEquals("timestamp-millis",
+ TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.CORRECT,
DataShape.MILLIS, false).get());
+ // Ambiguous / divergent / unaffected: no automatic suggestion.
+
assertFalse(TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.AMBIGUOUS,
DataShape.UNKNOWN, false).isPresent());
+
assertFalse(TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.UNAFFECTED,
DataShape.UNKNOWN, false).isPresent());
+
assertFalse(TimestampLogicalTypeClassifier.suggestedOverrideToken(Bucket.DIVERGENT,
DataShape.MICROS, false).isPresent());
+ }
+
+ @Test
+ public void testBucketsMillisSideSymmetry() {
+ // Mirror the micros cases in testBuckets() with the millis side. The
symmetric coverage guards
+ // against a future edit accidentally handling only one direction.
+ assertEquals(Bucket.CORRECT,
+ TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MILLIS,
TIMESTAMP_MILLIS, DataShape.MILLIS));
+ // 0.14.1 drift on the millis side: label millis, values micros.
+ assertEquals(Bucket.LEGACY_0X_BUG,
+ TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MILLIS,
TIMESTAMP_MILLIS, DataShape.MICROS));
+ // A millis-labeled column with unjudgeable value shape.
+ assertEquals(Bucket.AMBIGUOUS,
+ TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MILLIS,
TIMESTAMP_MILLIS, DataShape.UNKNOWN));
+ // Table + file disagree in the reverse direction — falls through to
DIVERGENT.
+ assertEquals(Bucket.DIVERGENT,
+ TimestampLogicalTypeClassifier.classifyBucket(TIMESTAMP_MILLIS,
TIMESTAMP_MICROS, DataShape.MILLIS));
+ // Dropped local-timestamp-millis: bare long everywhere, values millis.
Covers the 0.x drop of
+ // local-timestamp logical types, millis side.
+ assertEquals(Bucket.DROPPED_LOGICAL_TYPE,
+ TimestampLogicalTypeClassifier.classifyBucket(NONE, NONE,
DataShape.MILLIS));
+ }
+
+ @Test
+ public void testBucketsLocalTimestampVariants() {
+ // Local-timestamp variants must classify identically to their non-local
counterparts —
+ // the bug happens against the same three signals, only the resulting
override token differs.
+ assertEquals(Bucket.CORRECT,
+ TimestampLogicalTypeClassifier.classifyBucket(LOCAL_TIMESTAMP_MICROS,
LOCAL_TIMESTAMP_MICROS, DataShape.MICROS));
+ assertEquals(Bucket.CORRECT,
+ TimestampLogicalTypeClassifier.classifyBucket(LOCAL_TIMESTAMP_MILLIS,
LOCAL_TIMESTAMP_MILLIS, DataShape.MILLIS));
+ assertEquals(Bucket.LEGACY_0X_BUG,
+ TimestampLogicalTypeClassifier.classifyBucket(LOCAL_TIMESTAMP_MICROS,
LOCAL_TIMESTAMP_MICROS, DataShape.MILLIS));
+ assertEquals(Bucket.LEGACY_0X_BUG,
+ TimestampLogicalTypeClassifier.classifyBucket(LOCAL_TIMESTAMP_MILLIS,
LOCAL_TIMESTAMP_MILLIS, DataShape.MICROS));
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
index 22c1905a996a..234ef5a556b1 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
@@ -23,10 +23,11 @@ import
org.apache.hudi.HoodieSparkSqlWriter.{CANONICALIZE_SCHEMA, SQL_MERGE_INTO
import org.apache.hudi.common.config.{HoodieCommonConfig, HoodieConfig,
TypedProperties}
import org.apache.hudi.common.model.HoodieRecord
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaCompatibility,
HoodieSchemaUtils => HoodieCommonSchemaUtils}
-import org.apache.hudi.common.schema.internal.InternalSchema
+import org.apache.hudi.common.schema.internal.{InternalSchema, Type}
import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter
import org.apache.hudi.common.schema.internal.utils.AvroSchemaEvolutionUtils
import
org.apache.hudi.common.schema.internal.utils.AvroSchemaEvolutionUtils.reconcileSchemaRequirements
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils
import org.apache.hudi.common.table.{HoodieTableMetaClient,
TableSchemaResolver}
import org.apache.hudi.common.util.ConfigUtils
import org.apache.hudi.config.HoodieWriteConfig
@@ -141,10 +142,26 @@ object HoodieSchemaUtils {
InternalSchemaConverter.fixNullOrdering(sourceSchema)
}
+ // Parse the per-field timestamp overrides once and thread the parsed
map through the
+ // upfront guard and every downstream branch — reduces cognitive load
and matches the
+ // "single source of truth" theme of the config.
+ val timestampLogicalTypeOverrides =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+
opts.getOrElse(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key,
+ HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.defaultValue))
+
+ // Reconcile timestamp precision up front so every downstream branch
(including the
+ // non-reconcile default path, whose Avro compatibility check is
logical-type-blind) is
+ // guarded: an unverified micros/millis change throws here rather than
silently flipping the
+ // table on the next commit.
+ val precisionReconciledSourceSchema =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(
+ canonicalizedSourceSchema, latestTableSchema,
timestampLogicalTypeOverrides)
+
if (shouldReconcileSchema) {
- deduceWriterSchemaWithReconcile(sourceSchema,
canonicalizedSourceSchema, latestTableSchema, internalSchemaOpt, opts)
+ deduceWriterSchemaWithReconcile(sourceSchema,
precisionReconciledSourceSchema, latestTableSchema,
+ internalSchemaOpt, opts, timestampLogicalTypeOverrides)
} else {
- deduceWriterSchemaWithoutReconcile(sourceSchema,
canonicalizedSourceSchema, latestTableSchema, opts)
+ deduceWriterSchemaWithoutReconcile(sourceSchema,
precisionReconciledSourceSchema, latestTableSchema,
+ opts, timestampLogicalTypeOverrides)
}
}
}
@@ -157,7 +174,8 @@ object HoodieSchemaUtils {
private def deduceWriterSchemaWithoutReconcile(sourceSchema: HoodieSchema,
canonicalizedSourceSchema:
HoodieSchema,
latestTableSchema:
HoodieSchema,
- opts: Map[String, String]):
HoodieSchema = {
+ opts: Map[String, String],
+
timestampLogicalTypeOverrides: java.util.Map[String, Type]): HoodieSchema = {
// NOTE: In some cases we need to relax constraint of incoming dataset's
schema to be compatible
// w/ the table's one and allow schemas to diverge. This is required
in cases where
// partial updates will be performed (for ex, `MERGE INTO` Spark SQL
statement) and as such
@@ -173,7 +191,8 @@ object HoodieSchemaUtils {
if (!mergeIntoWrites && !shouldValidateSchemasCompatibility &&
!allowAutoEvolutionColumnDrop) {
// Default behaviour
val reconciledSchema = if (setNullForMissingColumns) {
- AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema,
latestTableSchema, setNullForMissingColumns)
+ AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema,
latestTableSchema,
+ setNullForMissingColumns, timestampLogicalTypeOverrides)
} else {
canonicalizedSourceSchema
}
@@ -199,13 +218,15 @@ object HoodieSchemaUtils {
canonicalizedSourceSchema:
HoodieSchema,
latestTableSchema: HoodieSchema,
internalSchemaOpt:
Option[InternalSchema],
- opts: Map[String, String]):
HoodieSchema = {
+ opts: Map[String, String],
+ timestampLogicalTypeOverrides:
java.util.Map[String, Type]): HoodieSchema = {
internalSchemaOpt match {
case Some(internalSchema) =>
// Apply schema evolution, by auto-merging write schema and read schema
val setNullForMissingColumns =
opts.getOrElse(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key(),
HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.defaultValue()).toBoolean
- val mergedInternalSchema =
AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema,
internalSchema, setNullForMissingColumns)
+ val mergedInternalSchema =
AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema,
internalSchema,
+ setNullForMissingColumns, timestampLogicalTypeOverrides)
val evolvedSchema =
InternalSchemaConverter.convert(mergedInternalSchema,
latestTableSchema.getFullName)
val shouldRemoveMetaDataFromInternalSchema =
sourceSchema.getFields.asScala.filter(f =>
f.name().equalsIgnoreCase(HoodieRecord.RECORD_KEY_METADATA_FIELD)).isEmpty
if (shouldRemoveMetaDataFromInternalSchema)
HoodieCommonSchemaUtils.removeMetadataFields(evolvedSchema) else evolvedSchema
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
index d7d046e2dff9..86165740dcbc 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
@@ -26,6 +26,7 @@ import org.apache.hudi.HoodieSparkUtils;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient;
import org.apache.hudi.common.config.DFSPropertiesConfiguration;
+import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.config.LockConfiguration;
@@ -79,6 +80,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.transaction.lock.InProcessLockProvider;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.SchemaCompatibilityException;
import org.apache.hudi.exception.TableNotFoundException;
import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
@@ -210,6 +212,11 @@ import static
org.junit.jupiter.params.provider.Arguments.arguments;
@Slf4j
public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
+ // Per-field verdict for the corrupt logical-repair fixtures: relabel
ts_millis to millis and
+ // attach the local-timestamp logical types that 0.x dropped. ts_micros is
already micros.
+ private static final String LOGICAL_REPAIR_TS_OVERRIDES =
+
"ts_millis:timestamp-millis,local_ts_millis:local-timestamp-millis,local_ts_micros:local-timestamp-micros";
+
private void addRecordMerger(HoodieRecordType type, List<String>
hoodieConfig) {
if (type == HoodieRecordType.SPARK) {
Map<String, String> opts = new HashMap<>();
@@ -967,6 +974,13 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
String schemaPath = zipOutput + "/schema.avsc";
cfg.configs.add(String.format(("%s=%s"),
"hoodie.streamer.schemaprovider.source.schema.file", schemaPath));
cfg.configs.add(String.format(("%s=%s"),
"hoodie.streamer.schemaprovider.target.schema.file", schemaPath));
+ // The v6/v8 col-stats fixture reuses the same trips_logical_types_json
corrupt schema as
+ // the logical-repair tests — 0.x collapsed ts_millis to
timestamp-micros and dropped the
+ // local-timestamp logical types entirely. Provide the same explicit
verdict so the guard
+ // in HoodieSchemaUtils.deduceWriterSchema authorizes the repair rather
than rejecting the
+ // unverified precision change.
+ cfg.configs.add(String.format(("%s=%s"),
+ HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key(),
LOGICAL_REPAIR_TS_OVERRIDES));
cfg.forceDisableCompaction = true;
cfg.sourceLimit = 100_000;
cfg.ignoreCheckpoint = "12345";
@@ -1109,8 +1123,19 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
}
@ParameterizedTest
- @CsvSource(value = {"SIX,AVRO,CLUSTER", "EIGHT,AVRO,CLUSTER",
"CURRENT,AVRO,NONE", "CURRENT,AVRO,CLUSTER", "CURRENT,SPARK,NONE",
"CURRENT,SPARK,CLUSTER"})
- void testCOWLogicalRepair(String tableVersion, String recordType, String
operation) throws Exception {
+ @CsvSource(value = {
+ // Repair succeeds when a per-field verdict is set, on the default
(non-reconcile) write path...
+ "SIX,AVRO,CLUSTER,false,true", "EIGHT,AVRO,CLUSTER,false,true",
+ "CURRENT,AVRO,NONE,false,true", "CURRENT,AVRO,CLUSTER,false,true",
+ "CURRENT,SPARK,NONE,false,true", "CURRENT,SPARK,CLUSTER,false,true",
+ // ...and on the reconcile path (setNullForMissingColumns=true).
+ "SIX,AVRO,CLUSTER,true,true", "EIGHT,AVRO,CLUSTER,true,true",
"CURRENT,AVRO,CLUSTER,true,true",
+ // Guard: with no verdict, the mislabeled timestamp/local-timestamp
columns must be rejected on
+ // the first sync, on both the reconcile path and the default path.
+ "SIX,AVRO,CLUSTER,true,false", "SIX,AVRO,CLUSTER,false,false"})
+ void testCOWLogicalRepair(String tableVersion, String recordType, String
operation,
+ boolean setNullForMissingColumns,
+ boolean setTimestampOverride) throws Exception {
TestMercifulJsonToRowConverterBase.timestampNTZCompatibility(() -> {
String dirName = "trips_logical_types_json_cow_write";
String dataPath = basePath + "/" + dirName;
@@ -1139,9 +1164,36 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
properties.setProperty("hoodie.parquet.small.file.limit", "-1");
properties.setProperty("hoodie.cleaner.commits.retained", "10");
properties.setProperty(HoodieWriteConfig.WRITE_TABLE_VERSION.key(),
tableVersionString);
+
properties.setProperty(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key(),
+ Boolean.toString(setNullForMissingColumns));
+ if (setTimestampOverride) {
+ // Per-field verdict authorizing the repair: relabel ts_millis to
millis and attach the
+ // local-timestamp logical types 0.x dropped. ts_micros stays micros
(no entry needed).
+
properties.setProperty(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key(),
+ LOGICAL_REPAIR_TS_OVERRIDES);
+ }
Option<TypedProperties> propt = Option.of(properties);
+ if (!setTimestampOverride) {
+ // No per-field verdict. The mislabeled timestamp/local-timestamp
columns must be rejected on
+ // the first sync rather than silently flipped, on both the reconcile
and default write paths.
+ // syncOnce wraps the guard's SchemaCompatibilityException in a
HoodieIngestionException, so
+ // walk the cause chain to assert on the underlying exception.
+ Throwable thrown = assertThrows(Exception.class,
+ () -> syncOnce(prepCfgForCowLogicalRepair(tableBasePath, "456"),
propt));
+ 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);
+ assertTrue(cause.getMessage().contains("column 'ts_millis'")
+ && cause.getMessage().contains("without an explicit"),
+ "Unexpected message: " + cause.getMessage());
+ return;
+ }
+
syncOnce(prepCfgForCowLogicalRepair(tableBasePath, "456"), propt);
inputDataPath =
getClass().getClassLoader().getResource("logical-repair/cow_write_updates/3").toURI().toString();
@@ -1189,11 +1241,17 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
}
@ParameterizedTest
- @CsvSource(value = {"SIX,AVRO,CLUSTER,AVRO", "EIGHT,AVRO,CLUSTER,AVRO",
- "CURRENT,AVRO,NONE,AVRO", "CURRENT,AVRO,CLUSTER,AVRO",
"CURRENT,AVRO,COMPACT,AVRO",
- "CURRENT,AVRO,NONE,PARQUET", "CURRENT,AVRO,CLUSTER,PARQUET",
"CURRENT,AVRO,COMPACT,PARQUET",
- "CURRENT,SPARK,NONE,PARQUET", "CURRENT,SPARK,CLUSTER,PARQUET",
"CURRENT,SPARK,COMPACT,PARQUET"})
- void testMORLogicalRepair(String tableVersion, String recordType, String
operation, String logBlockType) throws Exception {
+ @CsvSource(value = {"SIX,AVRO,CLUSTER,AVRO,false,true",
"EIGHT,AVRO,CLUSTER,AVRO,false,true",
+ "CURRENT,AVRO,NONE,AVRO,false,true",
"CURRENT,AVRO,CLUSTER,AVRO,false,true", "CURRENT,AVRO,COMPACT,AVRO,false,true",
+ "CURRENT,AVRO,NONE,PARQUET,false,true",
"CURRENT,AVRO,CLUSTER,PARQUET,false,true",
"CURRENT,AVRO,COMPACT,PARQUET,false,true",
+ "CURRENT,SPARK,NONE,PARQUET,false,true",
"CURRENT,SPARK,CLUSTER,PARQUET,false,true",
"CURRENT,SPARK,COMPACT,PARQUET,false,true",
+ // Variants that exercise the schema-reconcile path
(setNullForMissingColumns=true) with a verdict.
+ "SIX,AVRO,CLUSTER,AVRO,true,true", "EIGHT,AVRO,CLUSTER,AVRO,true,true",
"CURRENT,AVRO,CLUSTER,AVRO,true,true",
+ // Guard: with no verdict, the first sync must throw, on both the
reconcile and default paths.
+ "SIX,AVRO,CLUSTER,AVRO,true,false", "SIX,AVRO,CLUSTER,AVRO,false,false"})
+ void testMORLogicalRepair(String tableVersion, String recordType, String
operation, String logBlockType,
+ boolean setNullForMissingColumns,
+ boolean setTimestampOverride) throws Exception {
TestMercifulJsonToRowConverterBase.timestampNTZCompatibility(() -> {
String tableSuffix;
String logFormatValue;
@@ -1241,6 +1299,12 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
properties.setProperty("hoodie.cleaner.commits.retained", "10");
properties.setProperty(HoodieWriteConfig.WRITE_TABLE_VERSION.key(),
tableVersionString);
properties.setProperty(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(),
logFormatValue);
+
properties.setProperty(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key(),
+ Boolean.toString(setNullForMissingColumns));
+ if (setTimestampOverride) {
+
properties.setProperty(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key(),
+ LOGICAL_REPAIR_TS_OVERRIDES);
+ }
boolean disableCompaction;
if ("COMPACT".equals(operation)) {
@@ -1262,6 +1326,25 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
Option<TypedProperties> propt = Option.of(properties);
+ if (!setTimestampOverride) {
+ // No per-field verdict. The mislabeled timestamp/local-timestamp
columns must be rejected on
+ // the first sync rather than silently flipped, on both the reconcile
and default write paths.
+ // syncOnce wraps the guard's SchemaCompatibilityException in a
HoodieIngestionException, so
+ // walk the cause chain to assert on the underlying exception.
+ Throwable thrown = assertThrows(Exception.class,
+ () -> syncOnce(prepCfgForMorLogicalRepair(tableBasePath, dirName,
"123", disableCompaction), propt));
+ 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);
+ assertTrue(cause.getMessage().contains("column 'ts_millis'")
+ && cause.getMessage().contains("without an explicit"),
+ "Unexpected message: " + cause.getMessage());
+ return;
+ }
+
syncOnce(prepCfgForMorLogicalRepair(tableBasePath, dirName, "123",
disableCompaction), propt);
String prevTimezone =
sparkSession.conf().get("spark.sql.session.timeZone");
