hudi-agent commented on code in PR #20009:
URL: https://github.com/apache/hudi/pull/20009#discussion_r4056064975
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java:
##########
@@ -175,6 +176,10 @@ public boolean needsUpgrade(HoodieTableVersion
toWriteVersion) {
* @param instantTime current instant time that should not be touched.
*/
public void run(HoodieTableVersion toVersion, String instantTime) {
+ // Fail fast, before any rollback/compaction work is attempted, if the
target version exceeds the
+ // configured table version pin.
+ metaClient.getTableConfig().validateVersionPin(toVersion);
Review Comment:
🤖 This does fail before the rollback/compaction inside `run`, but the main
writer entry point `BaseHoodieWriteClient.tryUpgrade` (L1690–1702) calls
`needsUpgrade` and then eagerly rolls back all failed/inflight writes *before*
it ever reaches `run`, so a pinned table with a higher
`hoodie.write.table.version` will still get those rollbacks and only then hit
the pin. Would it make sense to also check the pin in
`needsUpgrade`/`needsUpgradeOrDowngrade` (or at the top of `tryUpgrade`) so the
writer really does no work first?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -793,9 +806,79 @@ public static String
getPartitionFieldWithoutKeyGenPartitionType(String partitio
* This function returns the hoodie.table.version from hoodie.properties
file.
*/
public static HoodieTableVersion getTableVersion(HoodieConfig config) {
- return contains(VERSION, config)
+ HoodieTableVersion version = contains(VERSION, config)
? HoodieTableVersion.fromVersionCode(config.getInt(VERSION))
: VERSION.defaultValue();
+ enforceVersionPin(config, version);
+ return version;
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.trim().isEmpty();
+ }
+
+ /**
+ * Resolves the configured {@link #MAX_ALLOWED_TABLE_VERSION} (falling back
to the JVM system property of the
+ * same key) into a version code, or {@link Option#empty()} if the pin is
disabled (unset, blank, or
+ * {@code UN_PINNED}).
+ */
+ private static Option<Integer> resolvePinnedVersionCode(HoodieConfig config)
{
+ String rawValue = contains(MAX_ALLOWED_TABLE_VERSION, config)
+ ? config.getString(MAX_ALLOWED_TABLE_VERSION)
+ : null;
+ // A blank table-level value counts as unset rather than as an explicit
opt-out, so that it cannot
+ // silently shadow a fleet-wide pin supplied through the system property.
+ if (isBlank(rawValue)) {
+ rawValue = System.getProperty(MAX_ALLOWED_TABLE_VERSION.key());
+ }
+ if (isBlank(rawValue) ||
MAX_ALLOWED_TABLE_VERSION.defaultValue().equalsIgnoreCase(rawValue.trim())) {
+ return Option.empty();
+ }
+ try {
+ int versionCode = Integer.parseInt(rawValue.trim());
+ if (versionCode < 0) {
+ throw new NumberFormatException("Table version pin must not be
negative: " + versionCode);
+ }
+ // Validates that the code corresponds to a recognized table version.
+ HoodieTableVersion.fromVersionCode(versionCode);
Review Comment:
🤖 If a fleet-wide pin (system property) is set to a version code that this
particular binary doesn't know yet (e.g. pin=11 while an older reader in the
same fleet only knows up to 10), this throws and the table becomes unloadable
for that reader, even though its actual version is fine. Since a pin above the
newest version this binary knows can never be exceeded, would it be safer to
treat that case as satisfied (or clamp to the max known version) rather than
fail?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -793,9 +806,79 @@ public static String
getPartitionFieldWithoutKeyGenPartitionType(String partitio
* This function returns the hoodie.table.version from hoodie.properties
file.
*/
public static HoodieTableVersion getTableVersion(HoodieConfig config) {
- return contains(VERSION, config)
+ HoodieTableVersion version = contains(VERSION, config)
? HoodieTableVersion.fromVersionCode(config.getInt(VERSION))
: VERSION.defaultValue();
+ enforceVersionPin(config, version);
+ return version;
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.trim().isEmpty();
+ }
+
+ /**
+ * Resolves the configured {@link #MAX_ALLOWED_TABLE_VERSION} (falling back
to the JVM system property of the
+ * same key) into a version code, or {@link Option#empty()} if the pin is
disabled (unset, blank, or
+ * {@code UN_PINNED}).
+ */
+ private static Option<Integer> resolvePinnedVersionCode(HoodieConfig config)
{
+ String rawValue = contains(MAX_ALLOWED_TABLE_VERSION, config)
+ ? config.getString(MAX_ALLOWED_TABLE_VERSION)
+ : null;
+ // A blank table-level value counts as unset rather than as an explicit
opt-out, so that it cannot
+ // silently shadow a fleet-wide pin supplied through the system property.
+ if (isBlank(rawValue)) {
+ rawValue = System.getProperty(MAX_ALLOWED_TABLE_VERSION.key());
+ }
+ if (isBlank(rawValue) ||
MAX_ALLOWED_TABLE_VERSION.defaultValue().equalsIgnoreCase(rawValue.trim())) {
+ return Option.empty();
+ }
+ try {
+ int versionCode = Integer.parseInt(rawValue.trim());
+ if (versionCode < 0) {
+ throw new NumberFormatException("Table version pin must not be
negative: " + versionCode);
+ }
+ // Validates that the code corresponds to a recognized table version.
+ HoodieTableVersion.fromVersionCode(versionCode);
+ return Option.of(versionCode);
+ } catch (NumberFormatException | HoodieException e) {
+ throw new HoodieTableVersionPinExceededException(
+ "Invalid value for " + MAX_ALLOWED_TABLE_VERSION.key() + ": '" +
rawValue
+ + "'. Must be '" + MAX_ALLOWED_TABLE_VERSION.defaultValue() + "'
or a recognized table version code.", e);
+ }
+ }
+
+ /**
+ * Throws {@link HoodieTableVersionPinExceededException} and emits a metric
when {@code version} exceeds the
+ * configured {@link #MAX_ALLOWED_TABLE_VERSION} ceiling. A version at or
below the pin is a no-op. Disabled
+ * entirely (no-op, no metric) when the pin is unset/{@code UN_PINNED}.
+ */
+ private static void enforceVersionPin(HoodieConfig config,
HoodieTableVersion version) {
+ Option<Integer> pinnedVersionCode = resolvePinnedVersionCode(config);
+ if (!pinnedVersionCode.isPresent() || version.versionCode() <=
pinnedVersionCode.get()) {
+ return;
+ }
+ String tableName = contains(NAME, config) ? config.getString(NAME) :
"unknown";
+ emitVersionPinExceededMetric(tableName);
+ throw new HoodieTableVersionPinExceededException(String.format(
+ "Table version %s (code %d) for table '%s' exceeds the pin %d
configured via '%s'. Refusing to proceed "
+ + "to avoid silently upgrading a table past its pinned version.",
+ version, version.versionCode(), tableName, pinnedVersionCode.get(),
MAX_ALLOWED_TABLE_VERSION.key()));
+ }
+
+ /**
+ * Emits a per-table counter metric via {@link Registry} when a table
version pin violation is detected. Failures
+ * emitting the metric are swallowed so they can never mask the {@link
HoodieTableVersionPinExceededException}
+ * thrown by the caller.
+ */
+ private static void emitVersionPinExceededMetric(String tableName) {
+ try {
+ Registry.getRegistryOfClass(tableName, "hoodie.table.version.pin",
LocalRegistry.class.getName())
Review Comment:
🤖 The description says any configured `MetricsReporter` picks this up, but
the only consumers of `Registry.getAllMetrics` are
`Metrics.flush()`/`shutdown()`, which exist only when a write client has
metrics enabled and run at commit/JVM-exit. On the read path (or a failed write
that never reaches a flush) the counter just accumulates in the static
`REGISTRY_MAP`. Is that the intended visibility, or should this be documented
as best-effort?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -180,6 +183,16 @@ public class HoodieTableConfig extends HoodieConfig {
+ " to identify what upgrade/downgrade paths happened on the table.
This is only configured "
+ "when the table is initially setup.");
+ public static final ConfigProperty<String> MAX_ALLOWED_TABLE_VERSION =
ConfigProperty
+ .key("hoodie.table.version.pinned")
+ .defaultValue("UN_PINNED")
+ .withDocumentation("Ceiling on hoodie.table.version: loading or writing
a table whose version exceeds "
+ + "this value throws immediately, so an accidental upgrade is caught
before it silently makes the "
+ + "table unreadable by older readers. A version at or below the pin
is unaffected. Also honored as "
+ + "a JVM system property of the same key (this config value takes
precedence over the system "
Review Comment:
🤖 Agree this is worth reconsidering. The system property is consulted on
every `getTableVersion()` call inside hudi-common, so one long-lived JVM
(shared Spark driver, Flink JM, timeline server) with the property set would
apply the pin to every table it touches, including reads of unrelated tables
above the pin. It also inverts the usual precedence here since any table's
`hoodie.properties` can set `UN_PINNED` to opt out of the fleet-wide pin. A
write-config option flowing through `HoodieWriteConfig` would fit the existing
pattern better.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -180,6 +183,16 @@ public class HoodieTableConfig extends HoodieConfig {
+ " to identify what upgrade/downgrade paths happened on the table.
This is only configured "
+ "when the table is initially setup.");
+ public static final ConfigProperty<String> MAX_ALLOWED_TABLE_VERSION =
ConfigProperty
+ .key("hoodie.table.version.pinned")
+ .defaultValue("UN_PINNED")
+ .withDocumentation("Ceiling on hoodie.table.version: loading or writing
a table whose version exceeds "
Review Comment:
🤖 +1, and there's a concrete reason to scope this to the upgrade path only:
`UpgradeDowngrade.run()` calls `getTableVersion()` to get `fromVersion`, and
that now throws whenever the current version is above the pin. So a table that
has already slipped past the pin can't be downgraded back — which is the
rollback scenario the PR description says it wants to protect. Gating only
`needsUpgrade`/the upgrade branch (and leaving `getTableVersion` alone) would
avoid that.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -180,6 +183,16 @@ public class HoodieTableConfig extends HoodieConfig {
+ " to identify what upgrade/downgrade paths happened on the table.
This is only configured "
+ "when the table is initially setup.");
+ public static final ConfigProperty<String> MAX_ALLOWED_TABLE_VERSION =
ConfigProperty
Review Comment:
🤖 nit: `MAX_ALLOWED_TABLE_VERSION` doesn't match the config key
`hoodie.table.version.pinned` — a reader searching for the pin field by name
would look for something like `TABLE_VERSION_PIN` first. Consider renaming the
constant to match the key/concept used elsewhere (e.g. `TABLE_VERSION_PINNED`).
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -793,9 +806,79 @@ public static String
getPartitionFieldWithoutKeyGenPartitionType(String partitio
* This function returns the hoodie.table.version from hoodie.properties
file.
*/
public static HoodieTableVersion getTableVersion(HoodieConfig config) {
- return contains(VERSION, config)
+ HoodieTableVersion version = contains(VERSION, config)
? HoodieTableVersion.fromVersionCode(config.getInt(VERSION))
: VERSION.defaultValue();
+ enforceVersionPin(config, version);
+ return version;
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.trim().isEmpty();
+ }
+
+ /**
+ * Resolves the configured {@link #MAX_ALLOWED_TABLE_VERSION} (falling back
to the JVM system property of the
+ * same key) into a version code, or {@link Option#empty()} if the pin is
disabled (unset, blank, or
+ * {@code UN_PINNED}).
+ */
+ private static Option<Integer> resolvePinnedVersionCode(HoodieConfig config)
{
+ String rawValue = contains(MAX_ALLOWED_TABLE_VERSION, config)
+ ? config.getString(MAX_ALLOWED_TABLE_VERSION)
+ : null;
+ // A blank table-level value counts as unset rather than as an explicit
opt-out, so that it cannot
+ // silently shadow a fleet-wide pin supplied through the system property.
+ if (isBlank(rawValue)) {
+ rawValue = System.getProperty(MAX_ALLOWED_TABLE_VERSION.key());
+ }
+ if (isBlank(rawValue) ||
MAX_ALLOWED_TABLE_VERSION.defaultValue().equalsIgnoreCase(rawValue.trim())) {
+ return Option.empty();
+ }
+ try {
+ int versionCode = Integer.parseInt(rawValue.trim());
+ if (versionCode < 0) {
+ throw new NumberFormatException("Table version pin must not be
negative: " + versionCode);
+ }
+ // Validates that the code corresponds to a recognized table version.
+ HoodieTableVersion.fromVersionCode(versionCode);
+ return Option.of(versionCode);
+ } catch (NumberFormatException | HoodieException e) {
+ throw new HoodieTableVersionPinExceededException(
+ "Invalid value for " + MAX_ALLOWED_TABLE_VERSION.key() + ": '" +
rawValue
+ + "'. Must be '" + MAX_ALLOWED_TABLE_VERSION.defaultValue() + "'
or a recognized table version code.", e);
+ }
+ }
+
+ /**
+ * Throws {@link HoodieTableVersionPinExceededException} and emits a metric
when {@code version} exceeds the
+ * configured {@link #MAX_ALLOWED_TABLE_VERSION} ceiling. A version at or
below the pin is a no-op. Disabled
+ * entirely (no-op, no metric) when the pin is unset/{@code UN_PINNED}.
+ */
+ private static void enforceVersionPin(HoodieConfig config,
HoodieTableVersion version) {
+ Option<Integer> pinnedVersionCode = resolvePinnedVersionCode(config);
+ if (!pinnedVersionCode.isPresent() || version.versionCode() <=
pinnedVersionCode.get()) {
+ return;
+ }
Review Comment:
🤖 nit: `resolvePinnedVersionCode` mixes resolution, validation, and
error-throwing for malformed values in one method — could you split
parsing/validation into a small helper so the happy-path resolution logic reads
more linearly?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
