Storages: introduce temporal min-max index for rough set filtering in columnar

AGENTS.md

@@ -135,6 +135,8 @@ When creating an **issue** or **pull request**, read the repo templates first an
 - **Issue:** `.github/ISSUE_TEMPLATE/` — pick the template that matches the request (bug, enhancement, feature, question, performance).
 - **PR:** `.github/pull_request_template.md` — use it for the PR title/body, checklists, and release note.
 
+When creating git commits, always use `-s` / `--signoff` so the commit message includes a `Signed-off-by` trailer (DCO).
+
 Typical fork workflow: branch from `upstream/master`, push to `origin`, open the PR against `pingcap/tiflash` (`master`).
 
 ## 📖 References

contrib/tiflash-columnar-hub/Cargo.toml

@@ -10,7 +10,7 @@ cloud_encryption = { git = "https://github.com/tidbcloud/cloud-storage-engine.gi
 keys = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine" }
 kvengine = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine" }
 kvenginepb = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine" }
-kvproto = { git = "https://github.com/pingcap/kvproto.git", rev = "683dad8fa3689deb243f2ff8ab5847c97af53e38" }
+kvproto = { git = "https://github.com/pingcap/kvproto.git", rev = "9327469" }
 pd_client = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine", default-features = false }
 prometheus = { version = "=0.13.0", features = ["nightly", "push"], default-features = true }
 security = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine", default-features = false }

contrib/tiflash-columnar-hub/hub-runtime/ffi/src/RaftStoreProxyFFI/ProxyFFI.h

@@ -251,7 +251,7 @@ struct CloudStorageEngineInterfaces {
   ColumnarReaderPtr (*fn_get_columnar_reader)(uint64_t, uint64_t, uint64_t,
                                               BaseBuffView, BaseBuffView,
                                               BaseBuffView, BaseBuffView,
-                                              BaseBuffView, BaseBuffView,
+                                              BaseBuffView, BaseBuffView, bool,
                                               RaftStoreProxyPtr);
   uint64_t (*fn_read_block)(ColumnarReaderPtr, uint64_t);
   RustStrWithView (*fn_read_handle)(ColumnarReaderPtr);

contrib/tiflash-columnar-hub/hub-runtime/src/cloud_helper.rs

@@ -527,6 +527,7 @@ impl CloudHelper {
         filter_conditions: Vec<tipb::Expr>,
         ann_query_info: tipb::AnnQueryInfo,
         fts_query_info: tipb::FtsQueryInfo,
+        enable_trim_minmax: bool,
     ) -> Result<CloudColumnarReaders, Error> {
         let dfs = self.dfs.clone();
         let pd_client = self.pd_client.clone();
@@ -537,12 +538,14 @@ impl CloudHelper {
         let ia_ctx = self.ia_ctx.clone();
         let http_client = self.http_client.clone();
         info!(
-            "make_columnar_reader, start_ts: {:?}, filter_conditions: {:?}, num_tables: {:?}",
+            "make_columnar_reader, start_ts: {:?}, filter_conditions: {:?}, num_tables: {:?}, enable_trim_minmax: {:?}",
             start_ts,
             filter_conditions,
-            tables.len()
+            tables.len(),
+            enable_trim_minmax
         );
-        let scan_ctx = TableScanCtx::new(table_scan, filter_conditions);
+        let scan_ctx =
+            TableScanCtx::new(table_scan, filter_conditions).with_enable_trim_minmax(enable_trim_minmax);
         let (tx, rx) = tikv_util::mpsc::bounded(1);
         let start = Instant::now_coarse();
         let vector_index_cache = self.vector_index_cache.clone();

contrib/tiflash-columnar-hub/hub-runtime/src/columnar_impls.rs

@@ -128,6 +128,7 @@ pub unsafe extern "C" fn ffi_make_columnar_reader(
     filter_conditions: BaseBuffView,
     ann_query_info: BaseBuffView,
     fts_query_info: BaseBuffView,
+    enable_trim_minmax: bool,
     hub_ptr: RaftStoreProxyPtr,
 ) -> ColumnarReaderPtr {
     let mut cols_pb = tipb::TableInfo::default();
@@ -218,6 +219,7 @@ pub unsafe extern "C" fn ffi_make_columnar_reader(
         filter_conditions_pb,
         ann_query_info_pb,
         fts_query_info_pb,
+        enable_trim_minmax,
     ) {
         Ok(reader) => (Box::into_raw(Box::new(reader)) as RawVoidPtr).into(),
         Err(err) => err.into(),

contrib/tiflash-columnar-hub/hub-runtime/src/interfaces.rs

@@ -393,7 +393,8 @@ pub mod root {
                     arg7: root::DB::BaseBuffView,
                     arg8: root::DB::BaseBuffView,
                     arg9: root::DB::BaseBuffView,
-                    arg10: root::DB::RaftStoreProxyPtr,
+                    arg10: bool,
+                    arg11: root::DB::RaftStoreProxyPtr,
                 ) -> root::DB::ColumnarReaderPtr,
             >,
             pub fn_read_block: ::std::option::Option<

dbms/src/Common/TiFlashMetrics.h

@@ -429,6 +429,38 @@ static_assert(RAFT_REGION_BIG_WRITE_THRES * 4 < RAFT_REGION_BIG_WRITE_MAX, "Inva
       "Bucketed histogram of rough set filter rate",                                                                                \
       Histogram,                                                                                                                    \
       F(type_dtfile_pack, {{"type", "dtfile_pack"}}, EqualWidthBuckets{0, 6, 20}))                                                  \
+    M(tiflash_storage_rough_set_pack_count,                                                                                         \
+      "Total number of packs before and after query rough set filtering",                                                           \
+      Counter,                                                                                                                      \
+      F(stage_query_input, {"stage", "query_input"}),                                                                               \
+      F(stage_query_filtered, {"stage", "query_filtered"}),                                                                         \
+      F(stage_query_remaining, {"stage", "query_remaining"}))                                                                       \
+    M(tiflash_storage_trim_minmax_select_count,                                                                                     \
+      "Total number of trim min-max selection results",                                                                             \
+      Counter,                                                                                                                      \
+      F(result_used, {"result", "used"}),                                                                                           \
+      F(result_fallback_disabled, {"result", "fallback_disabled"}),                                                                 \
+      F(result_fallback_non_meta_v2, {"result", "fallback_non_meta_v2"}),                                                           \
+      F(result_fallback_column_missing, {"result", "fallback_column_missing"}),                                                     \
+      F(result_fallback_no_meta, {"result", "fallback_no_meta"}),                                                                   \
+      F(result_fallback_unsupported_version, {"result", "fallback_unsupported_version"}),                                           \
+      F(result_fallback_metadata_mismatch, {"result", "fallback_metadata_mismatch"}),                                               \
+      F(result_fallback_index_missing, {"result", "fallback_index_missing"}),                                                       \
+      F(result_fallback_unsupported_expression, {"result", "fallback_unsupported_expression"}),                                     \
+      F(result_fallback_predicate_outside_range, {"result", "fallback_predicate_outside_range"}),                                   \
+      F(result_fallback_invalid_pack_marks, {"result", "fallback_invalid_pack_marks"}))                                             \
+    M(tiflash_storage_trim_minmax_rough_check_pack_count,                                                                           \
+      "Total number of packs by trim min-max rough check result",                                                                   \
+      Counter,                                                                                                                      \
+      F(result_none, {"result", "none"}),                                                                                           \
+      F(result_some, {"result", "some"}),                                                                                           \
+      F(result_all, {"result", "all"}),                                                                                             \
+      F(result_all_null, {"result", "all_null"}))                                                                                   \
+    M(tiflash_storage_trim_minmax_correction_pack_count,                                                                            \
+      "Total number of packs whose trim min-max rough check result is conservatively corrected",                                    \
+      Counter,                                                                                                                      \
+      F(type_none_to_some, {"type", "none_to_some"}),                                                                               \
+      F(type_all_to_some, {"type", "all_to_some"}))                                                                                 \
     M(tiflash_disaggregated_object_lock_request_count,                                                                              \
       "Total number of S3 object lock/delete request",                                                                              \
       Counter,                                                                                                                      \

dbms/src/Interpreters/Settings.h

@@ -177,6 +177,7 @@ struct Settings
     M(SettingFloat, dt_bg_gc_delta_delete_ratio_to_trigger_gc, 0.3, "Trigger segment's gc when the ratio of delta delete range to stable exceeds this ratio.")                                                                          \
     M(SettingBool, dt_enable_logical_split, false, "Enable logical split or not in DeltaTree Engine.")                                                                                                                                  \
     M(SettingBool, dt_enable_rough_set_filter, true, "Whether to parse where expression as Rough Set Index filter or not.")                                                                                                             \
+    M(SettingBool, dt_enable_trim_minmax, false, "Whether to generate and use trim min-max index for DATE/DATETIME/TIMESTAMP Rough Set filtering.")                                                                                    \
     M(SettingBool, dt_enable_relevant_place, false, "Enable relevant place or not in DeltaTree Engine.")                                                                                                                                \
     M(SettingBool, dt_enable_skippable_place, true, "Enable skippable place or not in DeltaTree Engine.")                                                                                                                               \
     M(SettingBool, dt_enable_stable_column_cache, true, "Enable column cache for StorageDeltaMerge.")                                                                                                                                   \

dbms/src/Storages/DeltaMerge/File/ColumnStat.h

@@ -22,6 +22,8 @@
 #include <Storages/FormatVersion.h>
 #include <Storages/KVStore/Types.h>
 
+#include <optional>
+
 namespace DB::DM
 {
 struct ColumnStat
@@ -44,6 +46,9 @@ struct ColumnStat
 
     std::vector<dtpb::DMFileIndexInfo> indexes{};
 
+    // Optional trim min-max metadata. Independent from `indexes` / local-index lifecycle.
+    std::optional<dtpb::TrimMinMaxIndexProps> trim_minmax_index{};
+
 #ifndef NDEBUG
     // This field is only used for testing
     String additional_data_for_test{};
@@ -71,6 +76,9 @@ struct ColumnStat
             pb_idx->CopyFrom(idx);
         }
 
+        if (trim_minmax_index.has_value())
+            *stat.mutable_trim_minmax_index() = *trim_minmax_index;
+
 #ifndef NDEBUG
         stat.set_additional_data_for_test(additional_data_for_test);
 #endif
@@ -131,6 +139,13 @@ struct ColumnStat
             indexes.emplace_back(pb_idx);
         }
 
+        // Soft-load only. Structural validation and fallback happen at selection time so a
+        // corrupt / unsupported trim meta never fails DMFile open.
+        if (proto.has_trim_minmax_index())
+            trim_minmax_index = proto.trim_minmax_index();
+        else
+            trim_minmax_index.reset();
+
 #ifndef NDEBUG
         additional_data_for_test = proto.additional_data_for_test();
 #endif
@@ -237,6 +252,7 @@ readText(ColumnStats & column_sats, DMFileFormat::Version ver, ReadBuffer & buf)
                 .serialized_bytes = serialized_bytes,
                 // ... here ignore some fields with default initializers
                 .indexes = {},
+                .trim_minmax_index = {},
 #ifndef NDEBUG
                 .additional_data_for_test = {},
 #endif

dbms/src/Storages/DeltaMerge/File/DMFile.cpp

@@ -192,6 +192,12 @@ String DMFile::colIndexCacheKey(const FileNameBase & file_name_base) const
     return colIndexPath(file_name_base);
 }
 
+String DMFile::colTrimIndexCacheKey(const FileNameBase & file_name_base) const
+{
+    // Distinct from ordinary `.idx` cache key; path already ends with `.trim.idx`.
+    return colTrimIndexPath(file_name_base);
+}
+
 String DMFile::colMarkCacheKey(const FileNameBase & file_name_base) const
 {
     return colMarkPath(file_name_base);
@@ -288,6 +294,11 @@ EncryptionPath DMFile::encryptionIndexPath(const FileNameBase & file_name_base)
     return EncryptionPath(encryptionBasePath(), file_name_base + details::INDEX_FILE_SUFFIX, keyspaceId());
 }
 
+EncryptionPath DMFile::encryptionTrimIndexPath(const FileNameBase & file_name_base) const
+{
+    return EncryptionPath(encryptionBasePath(), file_name_base + details::TRIM_INDEX_FILE_SUFFIX, keyspaceId());
+}
+
 EncryptionPath DMFile::encryptionMarkPath(const FileNameBase & file_name_base) const
 {
     return EncryptionPath(encryptionBasePath(), file_name_base + details::MARK_FILE_SUFFIX, keyspaceId());

dbms/src/Storages/DeltaMerge/File/DMFile.h

@@ -290,17 +290,23 @@ class DMFile : private boost::noncopyable
     {
         return subFilePath(colIndexFileName(file_name_base));
     }
+    String colTrimIndexPath(const FileNameBase & file_name_base) const
+    {
+        return subFilePath(colTrimIndexFileName(file_name_base));
+    }
     String colMarkPath(const FileNameBase & file_name_base) const
     {
         return subFilePath(colMarkFileName(file_name_base));
     }
 
     String colIndexCacheKey(const FileNameBase & file_name_base) const;
+    String colTrimIndexCacheKey(const FileNameBase & file_name_base) const;
     String colMarkCacheKey(const FileNameBase & file_name_base) const;
 
     String encryptionBasePath() const;
     EncryptionPath encryptionDataPath(const FileNameBase & file_name_base) const;
     EncryptionPath encryptionIndexPath(const FileNameBase & file_name_base) const;
+    EncryptionPath encryptionTrimIndexPath(const FileNameBase & file_name_base) const;
     EncryptionPath encryptionMarkPath(const FileNameBase & file_name_base) const;
 
     static FileNameBase getFileNameBase(ColId col_id, const IDataType::SubstreamPath & substream = {})

dbms/src/Storages/DeltaMerge/File/DMFileBlockInputStream.cpp

@@ -88,7 +88,9 @@ DMFileBlockInputStreamPtr DMFileBlockInputStreamBuilder::buildNoLocalIndex(
             rowkey_ranges,
             EMPTY_RS_OPERATOR,
             read_packs,
-            tracing_id);
+            tracing_id,
+            enable_trim_minmax,
+            read_tag);
     }
 
     DMFileReader reader(
@@ -137,6 +139,7 @@ SkippableBlockInputStreamPtr createSimpleBlockInputStream(
 DMFileBlockInputStreamBuilder & DMFileBlockInputStreamBuilder::setFromSettings(const Settings & settings)
 {
     enable_column_cache = settings.dt_enable_stable_column_cache;
+    enable_trim_minmax = settings.dt_enable_trim_minmax;
     max_read_buffer_size = settings.max_read_buffer_size;
     max_sharing_column_bytes_for_all = settings.dt_max_sharing_column_bytes_for_all;
     return *this;
@@ -210,7 +213,9 @@ SkippableBlockInputStreamPtr DMFileBlockInputStreamBuilder::buildForVectorIndex(
             rowkey_ranges,
             EMPTY_RS_OPERATOR,
             read_packs,
-            tracing_id);
+            tracing_id,
+            enable_trim_minmax,
+            read_tag);
     }
 
     DMFileReader rest_columns_reader(
@@ -296,7 +301,9 @@ SkippableBlockInputStreamPtr DMFileBlockInputStreamBuilder::buildForFullTextInde
             rowkey_ranges,
             EMPTY_RS_OPERATOR,
             read_packs,
-            tracing_id);
+            tracing_id,
+            enable_trim_minmax,
+            read_tag);
     }
 
     DMFileReader rest_columns_reader(

dbms/src/Storages/DeltaMerge/File/DMFileBlockInputStream.h

@@ -235,10 +235,10 @@ class DMFileBlockInputStreamBuilder
     FileProviderPtr file_provider;
 
     // clean read
-
     bool enable_handle_clean_read = false;
     bool is_fast_scan = false;
     bool enable_del_clean_read = false;
+    bool enable_trim_minmax = false;
     UInt64 max_data_version = std::numeric_limits<UInt64>::max();
     // packs filter (filter by pack index)
     IdSetPtr read_packs;

dbms/src/Storages/DeltaMerge/File/DMFileBlockOutputStream.cpp

@@ -31,7 +31,9 @@ DMFileBlockOutputStream::DMFileBlockOutputStream(
                 context.getSettingsRef().dt_compression_method,
                 context.getSettingsRef().dt_compression_level),
             context.getSettingsRef().min_compress_block_size,
-            context.getSettingsRef().max_compress_block_size})
+            context.getSettingsRef().max_compress_block_size,
+            context.getSettingsRef().dt_enable_trim_minmax,
+        })
 {}
 
-} // namespace DB::DM
+} // namespace DB::DM

dbms/src/Storages/DeltaMerge/File/DMFileMeta.cpp

@@ -25,6 +25,7 @@
 namespace DB::ErrorCodes
 {
 extern const int BAD_ARGUMENTS;
+extern const int LOGICAL_ERROR;
 } // namespace DB::ErrorCodes
 
 namespace DB::DM
@@ -53,6 +54,10 @@ void DMFileMeta::initializeIndices()
     directory.list(sub_files);
     for (const auto & name : sub_files)
     {
+        // Skip trim min-max files. Their names also end with `.idx` (`*.trim.idx`),
+        // but the prefix is not a plain ColId and must not enter column_indices.
+        if (endsWith(name, details::TRIM_INDEX_FILE_SUFFIX))
+            continue;
         if (endsWith(name, details::INDEX_FILE_SUFFIX))
         {
             column_indices.insert(
@@ -405,6 +410,14 @@ UInt64 DMFileMeta::getFileSize(ColId col_id, const String & filename) const
 {
     auto itr = column_stats.find(col_id);
     RUNTIME_CHECK(itr != column_stats.end(), col_id);
+    // Trim index size is only available from MergedSubFileInfo, never from ColumnStat.
+    if (endsWith(filename, details::TRIM_INDEX_FILE_SUFFIX))
+    {
+        throw Exception(
+            ErrorCodes::LOGICAL_ERROR,
+            "trim index size must be read from MergedSubFileInfo, filename={}",
+            filename);
+    }
     if (endsWith(filename, ".idx"))
     {
         return itr->second.index_bytes;