This is an automated email from the ASF dual-hosted git repository.
danny0405 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 d9e08f076370 feat(flink): add partitioned rli boostrap operator
(#19805)
d9e08f076370 is described below
commit d9e08f0763708dd937f211a6a883ba0b7e1939be
Author: Peter Huang <[email protected]>
AuthorDate: Thu Sep 3 05:14:16 2026 -0700
feat(flink): add partitioned rli boostrap operator (#19805)
* feat(flink): add partitioned rli boostrap operator
---
.../apache/hudi/configuration/FlinkOptions.java | 2 +-
.../apache/hudi/configuration/OptionsResolver.java | 13 ++
.../bootstrap/AbstractRLIBootstrapOperator.java | 86 +++++++++
.../sink/bootstrap/BootstrapOperatorFactory.java | 43 +++++
.../hudi/sink/bootstrap/RLIBootstrapOperator.java | 43 +----
.../bootstrap/TimeBoundedRLIBootstrapOperator.java | 194 +++++++++++++++++++
.../java/org/apache/hudi/sink/utils/Pipelines.java | 7 +-
.../org/apache/hudi/table/HoodieTableFactory.java | 1 -
.../java/org/apache/hudi/util/StreamerUtil.java | 20 ++
.../hudi/configuration/TestOptionsResolver.java | 31 +++
.../TestTimeBoundedRLIBootstrapOperator.java | 210 +++++++++++++++++++++
.../org/apache/hudi/sink/utils/TestPipelines.java | 47 +++++
.../apache/hudi/table/TestHoodieTableFactory.java | 61 ++++++
.../org/apache/hudi/utils/TestStreamerUtil.java | 22 +++
14 files changed, 736 insertions(+), 44 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
index 88662af61cb3..6ec2adea7116 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
@@ -374,7 +374,7 @@ public class FlinkOptions extends HoodieConfig {
public static final ConfigOption<Integer>
INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS = ConfigOptions
.key("index.rli.cache.rocksdb.bootstrap.days")
.intType()
- .defaultValue(7)
+ .defaultValue(-1)
.withDescription("Number of days of Partitioned Record Index to load
during bootstrap. Only partitions "
+ "within this window are pre-loaded; older partitions are loaded on
demand when updates are observed.");
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
index 07f31daa8616..f848b3ae4343 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
@@ -81,6 +81,9 @@ public class OptionsResolver {
// Value to override the default minimum file group count for global record
level index.
public static String GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_DEFAULT
= "8";
+ // Value of FlinkOptions#INDEX_RLI_BACKEND_TYPE that selects the local
RocksDB-based partitioned index cache.
+ private static final String ROCKSDB_INDEX_RLI_BACKEND_TYPE = "rocksdb";
+
/**
* Returns whether the current runtime mode is adaptive batch execution.
*/
@@ -262,6 +265,16 @@ public class OptionsResolver {
return indexType == HoodieIndex.IndexType.GLOBAL_RECORD_LEVEL_INDEX;
}
+ /**
+ * Returns whether the table uses partitioned record level index served by
the local RocksDB-based
+ * partitioned index cache, i.e. {@link FlinkOptions#INDEX_RLI_BACKEND_TYPE}
is configured as {@code rocksdb}.
+ */
+ public static boolean isTimeBoundedRLIBootstrapEnabled(Configuration conf) {
+ return conf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED)
+ &&
ROCKSDB_INDEX_RLI_BACKEND_TYPE.equalsIgnoreCase(conf.get(FlinkOptions.INDEX_RLI_BACKEND_TYPE))
+ && conf.get(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS) >
0;
+ }
+
/**
* Estimates the file group count to use for RLI partition of a new table.
*/
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/AbstractRLIBootstrapOperator.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/AbstractRLIBootstrapOperator.java
new file mode 100644
index 000000000000..1af2fbb6f251
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/AbstractRLIBootstrapOperator.java
@@ -0,0 +1,86 @@
+/*
+ * 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.sink.bootstrap;
+
+import org.apache.hudi.client.common.HoodieFlinkEngineContext;
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.util.StreamerUtil;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/**
+ * Base class for bootstrap operators that load record level index (RLI) data
from the metadata
+ * table, shared by {@link RLIBootstrapOperator} and {@link
TimeBoundedRLIBootstrapOperator}.
+ */
+@Slf4j
+public abstract class AbstractRLIBootstrapOperator
+ extends AbstractBootstrapOperator {
+
+ protected transient HoodieBackedTableMetadata tableMetadata;
+ protected transient long loadedCnt;
+
+ protected AbstractRLIBootstrapOperator(Configuration conf) {
+ super(conf);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeMetadataTable();
+ super.close();
+ }
+
+ // -------------------------------------------------------------------------
+ // Utilities
+ // -------------------------------------------------------------------------
+
+ protected HoodieBackedTableMetadata
createTableMetadata(HoodieTableMetaClient metaClient) {
+ return new HoodieBackedTableMetadata(
+ HoodieFlinkEngineContext.DEFAULT,
+ metaClient.getStorage(),
+ StreamerUtil.metadataConfig(conf),
+ conf.get(FlinkOptions.PATH));
+ }
+
+ protected void emitIndexRecord(String partitionPath, String recordKey,
HoodieRecordGlobalLocation location) {
+ output.collect(new StreamRecord<>(
+ new HoodieFlinkInternalRow(
+ recordKey,
+ partitionPath,
+ location.getFileId(),
+ String.valueOf(location.getInstantTime()))));
+ loadedCnt += 1;
+ }
+
+ protected void closeMetadataTable() {
+ if (tableMetadata != null) {
+ try {
+ tableMetadata.close();
+ } catch (Exception e) {
+ log.warn("Failed to close metadata table", e);
+ }
+ tableMetadata = null;
+ }
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperatorFactory.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperatorFactory.java
new file mode 100644
index 000000000000..34c82cf6c382
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperatorFactory.java
@@ -0,0 +1,43 @@
+/*
+ * 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.sink.bootstrap;
+
+import org.apache.hudi.configuration.OptionsResolver;
+
+import org.apache.flink.configuration.Configuration;
+
+/**
+ * Factory that resolves the concrete {@link AbstractBootstrapOperator}
implementation to use for
+ * the index bootstrap pipeline, keeping the pipeline construction agnostic of
the specific
+ * bootstrap operator selection logic.
+ */
+public final class BootstrapOperatorFactory {
+
+ private BootstrapOperatorFactory() {
+ }
+
+ public static AbstractBootstrapOperator createInstance(Configuration conf) {
+ if (OptionsResolver.isGlobalRecordLevelIndex(conf)) {
+ return new RLIBootstrapOperator(conf);
+ } else if (OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf)) {
+ return new TimeBoundedRLIBootstrapOperator(conf);
+ }
+ return new BootstrapOperator(conf);
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/RLIBootstrapOperator.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/RLIBootstrapOperator.java
index d0cb16a789e5..04f22478825a 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/RLIBootstrapOperator.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/RLIBootstrapOperator.java
@@ -18,23 +18,18 @@
package org.apache.hudi.sink.bootstrap;
-import org.apache.hudi.client.common.HoodieFlinkEngineContext;
-import org.apache.hudi.client.model.HoodieFlinkInternalRow;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.function.SerializableFunctionUnchecked;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.configuration.FlinkOptions;
-import org.apache.hudi.metadata.HoodieBackedTableMetadata;
import org.apache.hudi.util.StreamerUtil;
import org.apache.hudi.utils.RuntimeContextUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.StateInitializationContext;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import java.util.ArrayList;
import java.util.List;
@@ -51,10 +46,7 @@ import java.util.stream.Collectors;
*/
@Slf4j
public class RLIBootstrapOperator
- extends AbstractBootstrapOperator {
-
- private transient HoodieBackedTableMetadata tableMetadata;
- private transient long loadedCnt;
+ extends AbstractRLIBootstrapOperator {
public RLIBootstrapOperator(Configuration conf) {
super(conf);
@@ -64,21 +56,11 @@ public class RLIBootstrapOperator
public void initializeState(StateInitializationContext context) throws
Exception {
loadedCnt = 0;
HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf);
- this.tableMetadata = new HoodieBackedTableMetadata(
- HoodieFlinkEngineContext.DEFAULT,
- metaClient.getStorage(),
- StreamerUtil.metadataConfig(conf),
- conf.get(FlinkOptions.PATH));
+ this.tableMetadata = createTableMetadata(metaClient);
// Load RLI records
preLoadRLIRecords(metaClient.getTableConfig());
}
- @Override
- public void close() throws Exception {
- closeMetadataTable();
- super.close();
- }
-
// -------------------------------------------------------------------------
// Utilities
// -------------------------------------------------------------------------
@@ -133,24 +115,7 @@ public class RLIBootstrapOperator
return fileGroupIdx % parallelism == taskID;
}
- private void emitIndexRecord(String recordKey, HoodieRecordGlobalLocation
location) {
- output.collect(new StreamRecord<>(
- new HoodieFlinkInternalRow(
- recordKey,
- location.getPartitionPath(),
- location.getFileId(),
- String.valueOf(location.getInstantTime()))));
- loadedCnt += 1;
- }
-
- private void closeMetadataTable() {
- if (tableMetadata != null) {
- try {
- tableMetadata.close();
- } catch (Exception e) {
- log.warn("Failed to close metadata table", e);
- }
- tableMetadata = null;
- }
+ protected void emitIndexRecord(String recordKey, HoodieRecordGlobalLocation
location) {
+ emitIndexRecord(location.getPartitionPath(), recordKey, location);
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/TimeBoundedRLIBootstrapOperator.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/TimeBoundedRLIBootstrapOperator.java
new file mode 100644
index 000000000000..4d355e6e3e4d
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/TimeBoundedRLIBootstrapOperator.java
@@ -0,0 +1,194 @@
+/*
+ * 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.sink.bootstrap;
+
+import org.apache.hudi.common.data.HoodiePairData;
+import org.apache.hudi.common.function.SerializableFunctionUnchecked;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Functions;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.hash.BucketIndexUtil;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.RuntimeContextUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.StateInitializationContext;
+
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Bootstrap operator that preload of time bounded partitioned record level
index (RLI) data
+ * from the metadata table.
+ *
+ * <p>Only data table partitions that fall within the last {@link
FlinkOptions#INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS}
+ * days are eagerly preloaded; the partition path of each partition is parsed
as a date using
+ * {@link FlinkOptions#PARTITION_FORMAT} (default {@link
FlinkOptions#PARTITION_FORMAT_DAY}) to
+ * determine whether it falls inside the window. Partitions outside the
window, and partitions whose
+ * path cannot be parsed as a date, are skipped here and are expected to be
loaded on demand later.
+ *
+ * <p>Setting {@link FlinkOptions#INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS} to
{@code 0} disables preloading
+ * entirely, which is the expected fallback for non-temporal (non
date-partitioned) tables.
+ */
+@Slf4j
+public class TimeBoundedRLIBootstrapOperator
+ extends AbstractRLIBootstrapOperator {
+
+ private int parallelism;
+ private int taskID;
+ /**
+ * Functions for calculating the task partition to dispatch.
+ */
+ @VisibleForTesting
+ Functions.Function3<Integer, String, Integer, Integer> partitionIndexFunc;
+
+ public TimeBoundedRLIBootstrapOperator(Configuration conf) {
+ super(conf);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws
Exception {
+ loadedCnt = 0;
+ this.taskID =
RuntimeContextUtils.getIndexOfThisSubtask(getRuntimeContext());
+
+ int bootstrapDays =
conf.get(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS);
+ if (bootstrapDays <= 0) {
+ log.info("Skip preloading partitioned RLI records because bootstrap days
is configured as {}, taskId = {}",
+ bootstrapDays, taskID);
+ waitForBootstrapReady(taskID);
+ return;
+ }
+
+ HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf);
+ this.tableMetadata = createTableMetadata(metaClient);
+
+ this.parallelism =
RuntimeContextUtils.getNumberOfParallelSubtasks(getRuntimeContext());
+ this.partitionIndexFunc =
BucketIndexUtil.getPartitionIndexFunc(parallelism);
+ preLoadPartitionedRLIRecords(metaClient.getTableConfig(), bootstrapDays);
+ }
+
+ // -------------------------------------------------------------------------
+ // Utilities
+ // -------------------------------------------------------------------------
+
+ private void preLoadPartitionedRLIRecords(HoodieTableConfig tableConfig, int
bootstrapDays) {
+ if (!tableMetadata.enabled()) {
+ if (tableConfig.isMetadataTableAvailable()) {
+ throw new RuntimeException("Can not initialize the table metadata");
+ }
+ log.info("Skip preloading partitioned RLI records because table metadata
is not initialized, taskId = {}", taskID);
+ waitForBootstrapReady(taskID);
+ closeMetadataTable();
+ return;
+ }
+
+ if
(!tableConfig.isMetadataPartitionAvailable(MetadataPartitionType.RECORD_INDEX))
{
+ log.info("Skip preloading partitioned RLI records because record index
is not available yet, taskId = {}", taskID);
+ waitForBootstrapReady(taskID);
+ closeMetadataTable();
+ return;
+ }
+
+ Map<String, List<FileSlice>> partitionedFileGroups =
+
tableMetadata.getBucketizedFileGroupsForPartitionedRLI(MetadataPartitionType.RECORD_INDEX);
+ List<String> partitionsInWindow =
filterPartitionsInWindow(partitionedFileGroups.keySet(), bootstrapDays);
+
+ log.info("Start preloading partitioned RLI records from metadata table for
{}/{} partitions within the last {} days, "
+ + "taskId = {}, parallelism = {}",
+ partitionsInWindow.size(), partitionedFileGroups.size(),
bootstrapDays, taskID, parallelism);
+
+ long startTime = System.currentTimeMillis();
+ for (String partitionPath : partitionsInWindow) {
+ preLoadPartition(partitionPath,
partitionedFileGroups.get(partitionPath), taskID);
+ }
+ long costMs = System.currentTimeMillis() - startTime;
+ log.info("Finish preloading partitioned RLI records, total records: {},
cost: {} ms, taskId = {}", loadedCnt, costMs, taskID);
+
+ // Wait for other tasks to complete
+ waitForBootstrapReady(taskID);
+
+ // Cleanup resources
+ closeMetadataTable();
+ }
+
+ private void preLoadPartition(String partitionPath, List<FileSlice>
fileSlices, int taskID) {
+ List<FileSlice> filteredFileSlices = new ArrayList<>();
+ for (int i = 0; i < fileSlices.size(); i++) {
+ if (shouldLoadBucket(partitionPath, fileSlices.size(), i, taskID)) {
+ filteredFileSlices.add(fileSlices.get(i));
+ }
+ }
+ if (filteredFileSlices.isEmpty()) {
+ return;
+ }
+ log.info("Subtask: {} will preload partition {} from file groups: {},
total file groups: {}.",
+ taskID, partitionPath,
filteredFileSlices.stream().map(FileSlice::getFileId).collect(Collectors.joining(",")),
+ fileSlices.size());
+
+ // readRecordIndexLocations() discovers the full set of RLI file slices
internally and passes it to
+ // the filter; the filter here ignores that argument and substitutes the
file slices already scoped
+ // to this data partition, mirroring
RecordLevelIndexBackend#bootstrapPartition.
+ SerializableFunctionUnchecked<List<FileSlice>, List<FileSlice>>
fileSlicesFilter = fileSlicesToFilter -> filteredFileSlices;
+ HoodiePairData<String, HoodieRecordGlobalLocation> rliData =
tableMetadata.readRecordIndexLocations(fileSlicesFilter);
+ rliData.forEach(locationPair -> emitIndexRecord(partitionPath,
locationPair.getLeft(), locationPair.getRight()));
+ }
+
+ /**
+ * Determines if the given file group should be loaded by this task, using
the same
+ * partition-aware assignment as the write path (see {@link
BucketIndexUtil#getPartitionIndexFunc}),
+ * so that each file group is bootstrapped by the same task that owns it
during writes.
+ */
+ @VisibleForTesting
+ boolean shouldLoadBucket(String partitionPath, int fileGroupCount, int
fileGroupIdx, int taskID) {
+ return partitionIndexFunc.apply(fileGroupCount, partitionPath,
fileGroupIdx) == taskID;
+ }
+
+ /**
+ * Filters the data table partitions whose partition path can be parsed as a
date within the last
+ * {@code bootstrapDays} days, inclusive of today.
+ */
+ @VisibleForTesting
+ List<String> filterPartitionsInWindow(Iterable<String> partitionPaths, int
bootstrapDays) {
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern(
+
conf.getOptional(FlinkOptions.PARTITION_FORMAT).orElse(FlinkOptions.PARTITION_FORMAT_DAY));
+ boolean hiveStylePartitioning =
conf.get(FlinkOptions.HIVE_STYLE_PARTITIONING);
+ LocalDate today = conf.get(FlinkOptions.WRITE_UTC_TIMEZONE) ?
LocalDate.now(ZoneOffset.UTC) : LocalDate.now();
+ LocalDate cutoff = today.minusDays(bootstrapDays);
+
+ List<String> partitionsInWindow = new ArrayList<>();
+ for (String partitionPath : partitionPaths) {
+ LocalDate partitionDate = StreamerUtil.parsePartitionDate(partitionPath,
formatter, hiveStylePartitioning);
+ if (partitionDate != null && partitionDate.isAfter(cutoff) &&
!partitionDate.isAfter(today)) {
+ partitionsInWindow.add(partitionPath);
+ }
+ }
+ return partitionsInWindow;
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
index 7bf09e60234b..2410a3e18671 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
@@ -33,8 +33,8 @@ import org.apache.hudi.sink.CleanFunction;
import org.apache.hudi.sink.StreamWriteOperator;
import org.apache.hudi.sink.append.AppendWriteFunctions;
import org.apache.hudi.sink.append.AppendWriteOperator;
-import org.apache.hudi.sink.bootstrap.BootstrapOperator;
-import org.apache.hudi.sink.bootstrap.RLIBootstrapOperator;
+import org.apache.hudi.sink.bootstrap.AbstractBootstrapOperator;
+import org.apache.hudi.sink.bootstrap.BootstrapOperatorFactory;
import org.apache.hudi.sink.bootstrap.batch.BatchBootstrapOperator;
import org.apache.hudi.sink.bucket.BucketBulkInsertWriterHelper;
import org.apache.hudi.sink.bucket.BucketStreamWriteOperator;
@@ -427,11 +427,12 @@ public class Pipelines {
boolean isGlobalRLI = OptionsResolver.isGlobalRecordLevelIndex(conf);
if (conf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED) || (bounded &&
!isGlobalRLI)) {
+ AbstractBootstrapOperator bootstrapOperator =
BootstrapOperatorFactory.createInstance(conf);
dataStream1 = dataStream1
.transform(
"index_bootstrap",
new HoodieFlinkInternalRowTypeInfo(rowType),
- isGlobalRLI ? new RLIBootstrapOperator(conf) : new
BootstrapOperator(conf))
+ bootstrapOperator)
.setParallelism(conf.getOptional(FlinkOptions.INDEX_BOOTSTRAP_TASKS).orElse(dataStream1.getParallelism()))
.uid(opUID("index_bootstrap", conf));
((OneInputTransformation<?, ?>)
dataStream1.getTransformation()).setChainingStrategy(ChainingStrategy.ALWAYS);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
index 0d1437c0216c..554e75d1f4eb 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
@@ -493,7 +493,6 @@ public class HoodieTableFactory implements
DynamicTableSourceFactory, DynamicTab
conf.setString(HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key(),
"true");
conf.set(FlinkOptions.INDEX_GLOBAL_ENABLED, false);
conf.setString(HoodieMetadataConfig.STREAMING_WRITE_ENABLED.key(),
"true");
- conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, false);
if (!conf.contains(FlinkOptions.INDEX_RLI_WRITE_BUFFER_SIZE)) {
conf.set(FlinkOptions.INDEX_RLI_WRITE_BUFFER_SIZE,
OptionsResolver.getWriteBufferSizeInBytes(conf) / 1024 / 1024 / 4);
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
index 099da2c66eeb..4da31daddece 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
@@ -92,6 +92,9 @@ import org.apache.parquet.hadoop.ParquetFileWriter;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.StringReader;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
@@ -1025,4 +1028,21 @@ public class StreamerUtil {
}
return Option.empty();
}
+
+ public static LocalDate parsePartitionDate(String partitionPath,
DateTimeFormatter formatter, boolean hiveStylePartitioning) {
+ String dateValue = partitionPath;
+ if (hiveStylePartitioning) {
+ int idx = partitionPath.indexOf('=');
+ if (idx >= 0) {
+ dateValue = partitionPath.substring(idx + 1);
+ }
+ }
+ try {
+ return LocalDate.parse(dateValue, formatter);
+ } catch (DateTimeParseException e) {
+ log.warn("Skip preloading partition {} because its path cannot be parsed
as a date with format {}",
+ partitionPath, formatter, e);
+ return null;
+ }
+ }
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
index d0ebf5fba637..15ff8110dcbb 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
@@ -407,6 +407,37 @@ public class TestOptionsResolver {
OptionsResolver.getConflictResolutionStrategy(conf));
}
+ @Test
+ void testPartitionedRLIWithRocksDBBackend() {
+ // isTimeBoundedRLIBootstrapEnabled requires all three of:
INDEX_BOOTSTRAP_ENABLED explicitly
+ // turned on by the user (HoodieTableFactory no longer forces this for
RECORD_LEVEL_INDEX),
+ // a rocksdb-backed RLI cache, and a positive bootstrap-days window.
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true);
+ conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, 1);
+
+ // Wrong backend type.
+ conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "mdt");
+ assertFalse(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf));
+
+ // Backend type matches, case-insensitively.
+ conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "rocksdb");
+ assertTrue(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf));
+ conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "RocksDB");
+ assertTrue(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf));
+
+ // Bootstrap days must be positive.
+ conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, 0);
+ assertFalse(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf));
+ conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, 1);
+
+ // INDEX_BOOTSTRAP_ENABLED must be explicitly turned on, even with the
rest configured.
+ conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, false);
+ assertFalse(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf));
+ conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true);
+ }
+
@Test
void testWriteBufferSizingAndManagedMemory() {
Configuration conf = new Configuration();
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/TestTimeBoundedRLIBootstrapOperator.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/TestTimeBoundedRLIBootstrapOperator.java
new file mode 100644
index 000000000000..32771bd80c09
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/TestTimeBoundedRLIBootstrapOperator.java
@@ -0,0 +1,210 @@
+/*
+ * 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.sink.bootstrap;
+
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Functions;
+import org.apache.hudi.common.util.hash.BucketIndexUtil;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.TestConfigurations;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link TimeBoundedRLIBootstrapOperator}.
+ */
+public class TestTimeBoundedRLIBootstrapOperator {
+
+ @TempDir
+ File tempFile;
+
+ @Test
+ void testSkipPreloadWhenBootstrapDaysIsNonPositive() throws Exception {
+ Configuration conf = getTimeBoundedRLIConf();
+ conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, 0);
+ StreamerUtil.initTableIfNotExists(conf);
+
+ try (OneInputStreamOperatorTestHarness<HoodieFlinkInternalRow,
HoodieFlinkInternalRow> harness =
+ new OneInputStreamOperatorTestHarness<>(new
TimeBoundedRLIBootstrapOperator(conf), 1, 1, 0)) {
+ harness.open();
+
+ assertEquals(0, harness.getOutput().size());
+ }
+ }
+
+ @Test
+ void testSkipPreloadForFreshTableWithoutMetadataTable() throws Exception {
+ Configuration conf = getTimeBoundedRLIConf();
+ StreamerUtil.initTableIfNotExists(conf);
+
+ try (OneInputStreamOperatorTestHarness<HoodieFlinkInternalRow,
HoodieFlinkInternalRow> harness =
+ new OneInputStreamOperatorTestHarness<>(new
TimeBoundedRLIBootstrapOperator(conf), 1, 1, 0)) {
+ harness.open();
+
+ assertEquals(0, harness.getOutput().size());
+ }
+ }
+
+ @Test
+ void testFailFastWhenMetadataTableIsMarkedAvailableButCannotBeLoaded()
throws Exception {
+ Configuration conf = getTimeBoundedRLIConf();
+ HoodieTableMetaClient metaClient = StreamerUtil.initTableIfNotExists(conf);
+ metaClient.getTableConfig().setMetadataPartitionState(metaClient,
MetadataPartitionType.FILES.getPartitionPath(), true);
+
+ try (OneInputStreamOperatorTestHarness<HoodieFlinkInternalRow,
HoodieFlinkInternalRow> harness =
+ new OneInputStreamOperatorTestHarness<>(new
TimeBoundedRLIBootstrapOperator(conf), 1, 1, 0)) {
+ RuntimeException error = assertThrows(RuntimeException.class,
harness::open);
+
+ assertEquals("Can not initialize the table metadata",
error.getMessage());
+ }
+ }
+
+ @Test
+ void testFilterPartitionsInWindowWithDefaultDayFormat() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ TimeBoundedRLIBootstrapOperator operator = new
TimeBoundedRLIBootstrapOperator(conf);
+
+ DateTimeFormatter formatter =
DateTimeFormatter.ofPattern(FlinkOptions.PARTITION_FORMAT_DAY);
+ LocalDate today = LocalDate.now(ZoneOffset.UTC);
+ int bootstrapDays = 3;
+
+ String todayPartition = today.format(formatter);
+ String oneDayAgoPartition = today.minusDays(1).format(formatter);
+ String cutoffPartition = today.minusDays(bootstrapDays).format(formatter);
+ String justInsideWindowPartition = today.minusDays(bootstrapDays -
1).format(formatter);
+ String futurePartition = today.plusDays(1).format(formatter);
+ String unparsablePartition = "not-a-date";
+
+ List<String> partitionPaths = Arrays.asList(
+ todayPartition, oneDayAgoPartition, cutoffPartition,
justInsideWindowPartition, futurePartition, unparsablePartition);
+
+ List<String> result = operator.filterPartitionsInWindow(partitionPaths,
bootstrapDays);
+
+ assertEquals(3, result.size());
+ assertTrue(result.contains(todayPartition));
+ assertTrue(result.contains(oneDayAgoPartition));
+ assertTrue(result.contains(justInsideWindowPartition));
+ assertFalse(result.contains(cutoffPartition));
+ assertFalse(result.contains(futurePartition));
+ assertFalse(result.contains(unparsablePartition));
+ }
+
+ @Test
+ void testFilterPartitionsInWindowWithHiveStylePartitioning() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.HIVE_STYLE_PARTITIONING, true);
+ TimeBoundedRLIBootstrapOperator operator = new
TimeBoundedRLIBootstrapOperator(conf);
+
+ DateTimeFormatter formatter =
DateTimeFormatter.ofPattern(FlinkOptions.PARTITION_FORMAT_DAY);
+ LocalDate today = LocalDate.now(ZoneOffset.UTC);
+ int bootstrapDays = 3;
+
+ String todayPartition = "dt=" + today.format(formatter);
+ String outsideWindowPartition = "dt=" +
today.minusDays(bootstrapDays).format(formatter);
+
+ List<String> result =
operator.filterPartitionsInWindow(Arrays.asList(todayPartition,
outsideWindowPartition), bootstrapDays);
+
+ assertEquals(1, result.size());
+ assertTrue(result.contains(todayPartition));
+ }
+
+ @Test
+ void testFilterPartitionsInWindowWithCustomPartitionFormat() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.PARTITION_FORMAT,
FlinkOptions.PARTITION_FORMAT_DASHED_DAY);
+ TimeBoundedRLIBootstrapOperator operator = new
TimeBoundedRLIBootstrapOperator(conf);
+
+ DateTimeFormatter formatter =
DateTimeFormatter.ofPattern(FlinkOptions.PARTITION_FORMAT_DASHED_DAY);
+ LocalDate today = LocalDate.now(ZoneOffset.UTC);
+ int bootstrapDays = 1;
+
+ String todayPartition = today.format(formatter);
+ String outsideWindowPartition = today.minusDays(2).format(formatter);
+
+ List<String> result =
operator.filterPartitionsInWindow(Arrays.asList(todayPartition,
outsideWindowPartition), bootstrapDays);
+
+ assertEquals(1, result.size());
+ assertTrue(result.contains(todayPartition));
+ }
+
+ @Test
+ void testFilterPartitionsInWindowWithLocalTimezone() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.WRITE_UTC_TIMEZONE, false);
+ TimeBoundedRLIBootstrapOperator operator = new
TimeBoundedRLIBootstrapOperator(conf);
+
+ DateTimeFormatter formatter =
DateTimeFormatter.ofPattern(FlinkOptions.PARTITION_FORMAT_DAY);
+ LocalDate today = LocalDate.now();
+ int bootstrapDays = 1;
+
+ String todayPartition = today.format(formatter);
+
+ List<String> result =
operator.filterPartitionsInWindow(Arrays.asList(todayPartition), bootstrapDays);
+
+ assertEquals(1, result.size());
+ assertTrue(result.contains(todayPartition));
+ }
+
+ @Test
+ void testShouldLoadBucketMatchesPartitionIndexFunc() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ TimeBoundedRLIBootstrapOperator operator = new
TimeBoundedRLIBootstrapOperator(conf);
+
+ int parallelism = 4;
+ Functions.Function3<Integer, String, Integer, Integer> partitionIndexFunc
= BucketIndexUtil.getPartitionIndexFunc(parallelism);
+ operator.partitionIndexFunc = partitionIndexFunc;
+
+ String partitionPath = "20260101";
+ int fileGroupCount = 5;
+
+ for (int fileGroupIdx = 0; fileGroupIdx < fileGroupCount; fileGroupIdx++) {
+ int expectedTask = partitionIndexFunc.apply(fileGroupCount,
partitionPath, fileGroupIdx);
+ for (int taskID = 0; taskID < parallelism; taskID++) {
+ assertEquals(expectedTask == taskID,
operator.shouldLoadBucket(partitionPath, fileGroupCount, fileGroupIdx, taskID));
+ }
+ }
+ }
+
+ private Configuration getTimeBoundedRLIConf() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.METADATA_ENABLED, true);
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.GLOBAL_RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, 3);
+ return conf;
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestPipelines.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestPipelines.java
index 09340cd270ec..3b0244b5d343 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestPipelines.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestPipelines.java
@@ -25,6 +25,7 @@ import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.sink.bootstrap.TimeBoundedRLIBootstrapOperator;
import org.apache.hudi.sink.partitioner.GlobalRecordIndexPartitioner;
import org.apache.hudi.utils.TestConfigurations;
@@ -34,6 +35,8 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.runtime.partitioner.CustomPartitionerWrapper;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
@@ -48,6 +51,7 @@ import java.util.List;
import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -101,6 +105,49 @@ public class TestPipelines {
assertEquals(3, streaming.getParallelism());
}
+ @Test
+ void
testPartitionedRLIWithRocksDBBackendUsesPartitionedRLIBootstrapOperator() {
+ // HoodieTableFactory no longer forces INDEX_BOOTSTRAP_ENABLED to false
for RECORD_LEVEL_INDEX,
+ // so a user that wants time-bounded RLI bootstrap sets the flag
explicitly alongside the
+ // rocksdb backend config; set it here to exercise the same gate a
factory-built sink goes through.
+ Configuration conf = defaultConf();
+ conf.set(FlinkOptions.INDEX_GLOBAL_ENABLED, false);
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "rocksdb");
+ conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, 1);
+ conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true);
+ DataStream<RowData> input = rowDataInput();
+
+ DataStream<HoodieFlinkInternalRow> streaming =
+ Pipelines.bootstrap(conf, TestConfigurations.ROW_TYPE, input, false,
false);
+
+ assertEquals("index_bootstrap", streaming.getTransformation().getName());
+ assertInstanceOf(TimeBoundedRLIBootstrapOperator.class,
bootstrapOperator(streaming));
+ }
+
+ @Test
+ void
testPartitionedRLIWithRocksDBBackendSkipsBootstrapWhenNotExplicitlyEnabled() {
+ // With the forced false removed from HoodieTableFactory,
INDEX_BOOTSTRAP_ENABLED is now the
+ // single source of truth in Pipelines: rocksdb backend + bootstrap-days
config alone must not
+ // wire in a bootstrap operator unless the flag itself is turned on.
+ Configuration conf = defaultConf();
+ conf.set(FlinkOptions.INDEX_GLOBAL_ENABLED, false);
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "rocksdb");
+ conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, 1);
+ DataStream<RowData> input = rowDataInput();
+
+ DataStream<HoodieFlinkInternalRow> streaming =
+ Pipelines.bootstrap(conf, TestConfigurations.ROW_TYPE, input, false,
false);
+
+ assertNotEquals("index_bootstrap",
streaming.getTransformation().getName());
+ }
+
+ private Object bootstrapOperator(DataStream<HoodieFlinkInternalRow> stream) {
+ OneInputTransformation<?, ?> transformation = (OneInputTransformation<?,
?>) stream.getTransformation();
+ return ((SimpleOperatorFactory<?>)
transformation.getOperatorFactory()).getOperator();
+ }
+
@Test
void testWritePipelineOperatorGraphs() {
Configuration conf = defaultConf();
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
index 1bbfbb65418f..4400e580530b 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
@@ -68,8 +68,10 @@ import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Test cases for {@link HoodieTableFactory}.
@@ -849,6 +851,65 @@ public class TestHoodieTableFactory {
assertThat(globalRLIWithBootstrapResolvedConf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED),
is(true));
}
+ @Test
+ void testRecordLevelIndexBootstrapEnabledDefaultsToFalseWhenUnset() {
+ // HoodieTableFactory no longer forces INDEX_BOOTSTRAP_ENABLED for
RECORD_LEVEL_INDEX; when the
+ // user never sets it, the option's own default (false) should simply pass
through untouched.
+ Configuration rliConf = new Configuration(this.conf);
+ rliConf.set(FlinkOptions.OPERATION, "upsert");
+ rliConf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ rliConf.set(FlinkOptions.METADATA_ENABLED, true);
+ rliConf.set(FlinkOptions.INDEX_GLOBAL_ENABLED, false);
+
+ HoodieTableSink noBootstrapSink =
+ (HoodieTableSink) new
HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(rliConf));
+
assertThat(noBootstrapSink.getConf().get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED),
is(false));
+ }
+
+ @Test
+ void testFactoryBuiltSinkKeepsTimeBoundedRLIBootstrapEnabled() {
+ // The user is responsible for turning INDEX_BOOTSTRAP_ENABLED on for
time-bounded RLI
+ // bootstrap; HoodieTableFactory must pass that choice through unchanged
rather than
+ // overriding it, so the bootstrap pipeline stays reachable.
+ Configuration rliConf = new Configuration(this.conf);
+ rliConf.set(FlinkOptions.OPERATION, "upsert");
+ rliConf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ rliConf.set(FlinkOptions.METADATA_ENABLED, true);
+ rliConf.set(FlinkOptions.INDEX_GLOBAL_ENABLED, false);
+ rliConf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "rocksdb");
+ rliConf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, 1);
+ rliConf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true);
+
+ HoodieTableSink rliSink =
+ (HoodieTableSink) new
HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(rliConf));
+ Configuration rliResolvedConf = rliSink.getConf();
+
+ assertThat(rliResolvedConf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED),
is(true));
+
assertTrue(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(rliResolvedConf));
+ }
+
+ @Test
+ void
testFactoryDoesNotForceDisableExplicitBootstrapEnabledForRecordLevelIndex() {
+ // Documents the behavior change from this revision: previously the
factory always reset
+ // INDEX_BOOTSTRAP_ENABLED to false for RECORD_LEVEL_INDEX. Now a user-set
true survives even
+ // when the rocksdb time-bounded config isn't present, though
isTimeBoundedRLIBootstrapEnabled
+ // still requires the full rocksdb + bootstrap-days configuration to
select the time-bounded
+ // bootstrap operator.
+ Configuration rliConf = new Configuration(this.conf);
+ rliConf.set(FlinkOptions.OPERATION, "upsert");
+ rliConf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ rliConf.set(FlinkOptions.METADATA_ENABLED, true);
+ rliConf.set(FlinkOptions.INDEX_GLOBAL_ENABLED, false);
+ rliConf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true);
+
+ HoodieTableSink rliSink =
+ (HoodieTableSink) new
HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(rliConf));
+ Configuration rliResolvedConf = rliSink.getConf();
+
+ assertThat(rliResolvedConf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED),
is(true));
+
assertFalse(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(rliResolvedConf));
+ }
+
@Test
void testLanceFormatSupportedForFlinkTables() {
Configuration lanceConf = new Configuration();
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
index 3c77c51e5d73..becc48b64786 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
@@ -66,6 +66,8 @@ import org.mockito.Mockito;
import java.io.File;
import java.io.IOException;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -360,6 +362,26 @@ class TestStreamerUtil {
assertTrue(StreamerUtil.partitionExists(tempFile.getAbsolutePath(),
"dt=2026-08-06", hadoopConf));
}
+ @Test
+ void testParsePartitionDate() {
+ DateTimeFormatter dayFormatter =
DateTimeFormatter.ofPattern(FlinkOptions.PARTITION_FORMAT_DAY);
+ assertEquals(LocalDate.of(2026, 8, 6),
StreamerUtil.parsePartitionDate("20260806", dayFormatter, false));
+
+ DateTimeFormatter dashedDayFormatter =
DateTimeFormatter.ofPattern(FlinkOptions.PARTITION_FORMAT_DASHED_DAY);
+ assertEquals(LocalDate.of(2026, 8, 6),
StreamerUtil.parsePartitionDate("2026-08-06", dashedDayFormatter, false));
+
+ assertEquals(LocalDate.of(2026, 8, 6),
StreamerUtil.parsePartitionDate("dt=20260806", dayFormatter, true));
+
+ // hiveStylePartitioning=false must not strip the "dt=" prefix, so parsing
fails.
+ assertNull(StreamerUtil.parsePartitionDate("dt=20260806", dayFormatter,
false));
+
+ // no '=' present, hive-style parsing falls back to the raw path.
+ assertEquals(LocalDate.of(2026, 8, 6),
StreamerUtil.parsePartitionDate("20260806", dayFormatter, true));
+
+ assertNull(StreamerUtil.parsePartitionDate("not-a-date", dayFormatter,
false));
+ assertNull(StreamerUtil.parsePartitionDate("2026-08-06", dayFormatter,
false));
+ }
+
@Test
void testOrderingFieldAndKeyGeneratorValidation() {
Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
