[WIP]

dbms/src/Interpreters/Settings.h

@@ -177,6 +177,8 @@ 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_write, false, "Whether to write trim min-max index for DATE/DATETIME/TIMESTAMP columns in new DMFiles.")                                                                                        \
+    M(SettingBool, dt_enable_trim_minmax_read, false, "Whether to 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

@@ -14,6 +14,8 @@
 
 #pragma once
 
+#include <optional>
+
 #include <DataTypes/DataTypeFactory.h>
 #include <IO/ReadHelpers.h>
 #include <IO/WriteHelpers.h>
@@ -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/DMFileBlockOutputStream.cpp

@@ -31,7 +31,8 @@ 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_write})
 {}
 
 } // 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;

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

@@ -109,7 +109,7 @@ void DMFileMetaV2::parseColumnStat(std::string_view buffer)
         ColumnStat stat;
         stat.parseFromBuffer(rbuf);
         // Do not overwrite the ColumnStat if already exist, it may
-        // created by `ExteandColumnStat`
+        // created by `ExtendColumnStat`
         column_stats.emplace(stat.col_id, std::move(stat));
     }
 }
@@ -181,7 +181,7 @@ void DMFileMetaV2::finalize(
         writeSLPackStatToBuffer(tmp_buffer),
         writeSLPackPropertyToBuffer(tmp_buffer),
         writeExtendColumnStatToBuffer(tmp_buffer),
-        writeMergedSubFilePosotionsToBuffer(tmp_buffer),
+        writeMergedSubFilePositionsToBuffer(tmp_buffer),
     };
     writePODBinary(meta_block_handles, tmp_buffer);
     writeIntBinary(static_cast<UInt64>(meta_block_handles.size()), tmp_buffer);
@@ -255,7 +255,7 @@ DMFileMeta::BlockHandle DMFileMetaV2::writeExtendColumnStatToBuffer(WriteBuffer
     return BlockHandle{BlockType::ExtendColumnStat, offset, buffer.count() - offset};
 }
 
-DMFileMeta::BlockHandle DMFileMetaV2::writeMergedSubFilePosotionsToBuffer(WriteBuffer & buffer)
+DMFileMeta::BlockHandle DMFileMetaV2::writeMergedSubFilePositionsToBuffer(WriteBuffer & buffer)
 {
     auto offset = buffer.count();
 

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

@@ -125,7 +125,7 @@ class DMFileMetaV2 : public DMFileMeta
     BlockHandle writeSLPackPropertyToBuffer(WriteBuffer & buffer) const;
     BlockHandle writeColumnStatToBuffer(WriteBuffer & buffer);
     BlockHandle writeExtendColumnStatToBuffer(WriteBuffer & buffer);
-    BlockHandle writeMergedSubFilePosotionsToBuffer(WriteBuffer & buffer);
+    BlockHandle writeMergedSubFilePositionsToBuffer(WriteBuffer & buffer);
 
     // read
     void parse(std::string_view buffer);

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

@@ -17,10 +17,14 @@
 #include <Common/TiFlashMetrics.h>
 #include <DataTypes/IDataType.h>
 #include <IO/FileProvider/ChecksumReadBufferBuilder.h>
+#include <Storages/DeltaMerge/File/DMFileMetaV2.h>
 #include <Storages/DeltaMerge/File/DMFilePackFilter.h>
+#include <Storages/DeltaMerge/File/DMFileUtil.h>
+#include <Storages/DeltaMerge/Filter/DateQueryDomain.h>
 #include <Storages/DeltaMerge/Filter/FilterHelper.h>
 #include <Storages/DeltaMerge/Filter/RSOperator.h>
 #include <Storages/DeltaMerge/Index/RSResult.h>
+#include <Storages/DeltaMerge/Index/TrimMinMaxIndex.h>
 #include <Storages/DeltaMerge/RowKeyRange.h>
 #include <Storages/DeltaMerge/ScanContext.h>
 #include <Storages/S3/S3RandomAccessFile.h>
@@ -126,12 +130,9 @@ DMFilePackFilterResultPtr DMFilePackFilter::load()
     /// Check packs by filter in where clause
     if (filter)
     {
-        // Load index based on filter.
-        ColIds ids = filter->getColumnIDs();
-        for (const auto & id : ids)
-        {
-            tryLoadIndex(result.param, id);
-        }
+        // Load index based on filter requests (normal and/or PreferTrim).
+        for (const auto & request : filter->getIndexRequests())
+            tryLoadIndexByRequest(result.param, request);
 
         const auto check_results = filter->roughCheck(0, pack_count, result.param);
         std::transform(
@@ -377,6 +378,107 @@ void DMFilePackFilter::tryLoadIndex(RSCheckParam & param, ColId col_id)
     loadIndex(param.indexes, dmfile, file_provider, index_cache, set_cache_if_miss, col_id, read_limiter, scan_context);
 }
 
+void DMFilePackFilter::tryLoadIndexByRequest(RSCheckParam & param, const RSIndexRequest & request)
+{
+    if (request.preferred_kind == RSIndexKind::PreferTrim && request.query_domain.has_value())
+    {
+        if (tryLoadTrimIndex(param, request.col_id, *request.query_domain))
+            return;
+    }
+    tryLoadIndex(param, request.col_id);
+}
+
+bool DMFilePackFilter::tryLoadTrimIndex(RSCheckParam & param, ColId col_id, const DateQueryDomain & query_domain)
+{
+    if (param.trim_indexes.count(col_id))
+        return true;
+
+    if (!enable_trim_minmax_read || !dmfile->useMetaV2())
+        return false;
+
+    if (!dmfile->isColumnExist(col_id))
+        return false;
+
+    const auto & col_stat = dmfile->getColumnStat(col_id);
+    const auto * dmfile_meta = typeid_cast<const DMFileMetaV2 *>(dmfile->meta.get());
+    if (!dmfile_meta)
+        return false;
+
+    const auto file_name_base = DMFile::getFileNameBase(col_id);
+    const auto trim_fname = colTrimIndexFileName(file_name_base);
+
+    TrimMinMaxIndexMeta meta;
+    auto reason = TrimMinMax::trySelectTrimMeta(
+        /*read_enabled*/ true,
+        col_stat.trim_minmax_index,
+        *col_stat.type,
+        dmfile->getPacks(),
+        dmfile_meta->merged_sub_file_infos,
+        trim_fname,
+        &meta);
+    if (reason != TrimMinMaxFallbackReason::None)
+        return false;
+
+    if (!query_domain.isTrimEligible(meta.lower_bound, meta.upper_bound))
+        return false;
+
+    auto load_trim = [&]() -> MinMaxIndexPtr {
+        const auto file_path = dmfile->meta->mergedPath(meta.merged_file_number);
+        const auto offset = meta.merged_file_offset;
+        const auto data_size = meta.file_size;
+
+        auto index_guard = S3::S3RandomAccessFile::setReadFileInfo({
+            .size = dmfile->getReadFileSize(col_id, trim_fname),
+            .scan_context = scan_context,
+        });
+
+        auto buffer = ReadBufferFromRandomAccessFileBuilder::build(
+            file_provider,
+            file_path,
+            dmfile_meta->encryptionMergedPath(meta.merged_file_number),
+            std::min(data_size, dmfile->getConfiguration()->getChecksumFrameLength()),
+            read_limiter);
+        auto ret = buffer.seek(offset);
+        RUNTIME_CHECK_MSG(
+            ret >= 0,
+            "Failed to seek in merged file for trim index, ret={} file_path={} offset={}",
+            ret,
+            file_path,
+            offset);
+
+        String raw_data(data_size, '\0');
+        buffer.read(reinterpret_cast<char *>(raw_data.data()), data_size);
+
+        auto buf = ChecksumReadBufferBuilder::build(
+            std::move(raw_data),
+            file_path,
+            dmfile->getConfiguration()->getChecksumAlgorithm(),
+            dmfile->getConfiguration()->getChecksumFrameLength());
+
+        auto header_size = dmfile->getConfiguration()->getChecksumHeaderLength();
+        auto frame_total_size = dmfile->getConfiguration()->getChecksumFrameLength() + header_size;
+        auto frame_count = data_size / frame_total_size + (data_size % frame_total_size != 0);
+        return MinMaxIndex::read(*col_stat.type, *buf, data_size - header_size * frame_count);
+    };
+
+    MinMaxIndexPtr minmax_index;
+    if (index_cache && set_cache_if_miss)
+        minmax_index = index_cache->getOrSet(dmfile->colTrimIndexCacheKey(file_name_base), load_trim);
+    else
+    {
+        if (index_cache)
+            minmax_index = index_cache->get(dmfile->colTrimIndexCacheKey(file_name_base));
+        if (minmax_index == nullptr)
+            minmax_index = load_trim();
+    }
+
+    if (!minmax_index || !TrimMinMax::validateTrimPackMarks(minmax_index->packMarks(), meta.pack_count))
+        return false;
+
+    param.trim_indexes.emplace(col_id, TrimRSIndex{.type = col_stat.type, .minmax = minmax_index, .meta = meta});
+    return true;
+}
+
 std::pair<std::vector<DMFilePackFilter::Range>, DMFilePackFilterResults> DMFilePackFilter::getSkippedRangeAndFilter(
     const DMContext & dm_context,
     const DMFiles & dmfiles,

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

@@ -72,7 +72,8 @@ class DMFilePackFilter
             dm_context.global_context.getFileProvider(),
             dm_context.global_context.getReadLimiter(),
             dm_context.scan_context,
-            dm_context.tracing_id);
+            dm_context.tracing_id,
+            dm_context.global_context.getSettingsRef().dt_enable_trim_minmax_read);
         return pack_filter.load();
     }
 
@@ -86,7 +87,8 @@ class DMFilePackFilter
         const RowKeyRanges & rowkey_ranges,
         const RSOperatorPtr & filter,
         const IdSetPtr & read_packs,
-        const String & tracing_id)
+        const String & tracing_id,
+        bool enable_trim_minmax_read = false)
     {
         DMFilePackFilter pack_filter(
             dmfile,
@@ -98,7 +100,8 @@ class DMFilePackFilter
             file_provider_,
             read_limiter_,
             scan_context,
-            tracing_id);
+            tracing_id,
+            enable_trim_minmax_read);
         return pack_filter.load();
     }
 
@@ -170,7 +173,8 @@ class DMFilePackFilter
         const FileProviderPtr & file_provider_,
         const ReadLimiterPtr & read_limiter_,
         const ScanContextPtr & scan_context_,
-        const String & tracing_id)
+        const String & tracing_id,
+        bool enable_trim_minmax_read_)
         : dmfile(dmfile_)
         , index_cache(index_cache_)
         , set_cache_if_miss(set_cache_if_miss_)
@@ -181,6 +185,7 @@ class DMFilePackFilter
         , scan_context(scan_context_)
         , log(Logger::get(tracing_id))
         , read_limiter(read_limiter_)
+        , enable_trim_minmax_read(enable_trim_minmax_read_)
     {}
 
     DMFilePackFilterResultPtr load();
@@ -196,6 +201,8 @@ class DMFilePackFilter
         const ScanContextPtr & scan_context);
 
     void tryLoadIndex(RSCheckParam & param, ColId col_id);
+    void tryLoadIndexByRequest(RSCheckParam & param, const RSIndexRequest & request);
+    bool tryLoadTrimIndex(RSCheckParam & param, ColId col_id, const DateQueryDomain & query_domain);
 
 private:
     DMFilePtr dmfile;
@@ -211,6 +218,7 @@ class DMFilePackFilter
 
     LoggerPtr log;
     ReadLimiterPtr read_limiter;
+    bool enable_trim_minmax_read = false;
 };
 
 } // namespace DB::DM