Feature/tsfile dataframe runtime

.github/workflows/wheels.yml

@@ -96,6 +96,7 @@ jobs:
           ./mvnw -Pwith-cpp clean package \
             -DskipTests -Dspotless.check.skip=true -Dspotless.apply.skip=true \
             -Dbuild.test=OFF \
+            -Dtsfile.dependency.source=BUNDLED \
             -Denable.lzma2=ON \
             -Dcmake.args="-DCMAKE_OSX_DEPLOYMENT_TARGET=12.0"
           otool -l cpp/target/build/lib/libtsfile*.dylib | grep -A2 LC_VERSION_MIN_MACOSX || true
@@ -138,6 +139,7 @@ jobs:
             chmod +x mvnw || true
             ./mvnw -Pwith-cpp clean package \
               -DskipTests -Dbuild.test=OFF \
+              -Dtsfile.dependency.source=BUNDLED \
               -Denable.lzma2=ON \
               -Dspotless.check.skip=true -Dspotless.apply.skip=true
             test -d cpp/target/build/lib && test -d cpp/target/build/include
@@ -205,6 +207,7 @@ jobs:
           chmod +x mvnw || true
           ./mvnw -Pwith-cpp clean package \
             -DskipTests -Dbuild.test=OFF \
+            -Dtsfile.dependency.source=BUNDLED \
             -Denable.lzma2=ON \
             -Dspotless.check.skip=true -Dspotless.apply.skip=true
           test -d cpp/target/build/lib

.gitignore

@@ -32,6 +32,8 @@ python/tsfile/*dll*
 python/tsfile/*dylib*
 python/tsfile/*.h
 python/tsfile/*.cpp
+python/tsfile/**/*.cpp
+python/tsfile/**/*so*
 python/data
 python/venv/*
 python/tests/__pycache__/*

cpp/cmake/ANTLR4Dependency.cmake

@@ -19,7 +19,10 @@ under the License.
 
 set(TSFILE_ANTLR4_MIN_VERSION "4.9.3")
 set(TSFILE_ANTLR4_BUNDLED_VERSION "4.9.3")
-set(TSFILE_ANTLR4_NEXT_INCOMPATIBLE_VERSION "5.0.0")
+# The C++ runtime shipped by ANTLR4 4.13 and newer requires C++17. TsFile's
+# public C++ baseline remains C++11, so those system runtimes must not be
+# selected for the generated 4.9.3 parser.
+set(TSFILE_ANTLR4_NEXT_INCOMPATIBLE_VERSION "4.13.0")
 set(TSFILE_ANTLR4_SYSTEM_INCLUDE_DIR "")
 set(_TSFILE_SYSTEM_ANTLR4_FOUND FALSE)
 

cpp/cmake/tests/ANTLR4DependencyTest.cmake

@@ -59,6 +59,8 @@ function(_tsfile_run_antlr4_case NAME POLICY EXPECTED_SOURCE EXPECT_SUCCESS ROOT
                     "-DCMAKE_PREFIX_PATH=${ROOT}"
                     -DCMAKE_FIND_USE_PACKAGE_REGISTRY=FALSE
                     -DCMAKE_FIND_USE_SYSTEM_PACKAGE_REGISTRY=FALSE
+                    -DCMAKE_FIND_USE_CMAKE_SYSTEM_PATH=FALSE
+                    -DCMAKE_FIND_USE_SYSTEM_ENVIRONMENT_PATH=FALSE
                     -S "${_TSFILE_FIXTURE_SOURCE}"
                     -B "${_TSFILE_CASE_BINARY}"
             RESULT_VARIABLE _TSFILE_RESULT

cpp/cmake/tests/TestGeneratorArguments.cmake

@@ -22,6 +22,10 @@ if (TEST_CMAKE_GENERATOR)
     list(APPEND _TSFILE_TEST_GENERATOR_ARGUMENTS
             -G "${TEST_CMAKE_GENERATOR}")
 endif ()
+if (TEST_CMAKE_MAKE_PROGRAM)
+    list(APPEND _TSFILE_TEST_GENERATOR_ARGUMENTS
+            "-DCMAKE_MAKE_PROGRAM=${TEST_CMAKE_MAKE_PROGRAM}")
+endif ()
 if (TEST_CMAKE_GENERATOR_PLATFORM)
     list(APPEND _TSFILE_TEST_GENERATOR_ARGUMENTS
             -A "${TEST_CMAKE_GENERATOR_PLATFORM}")

cpp/src/CMakeLists.txt

@@ -72,6 +72,7 @@ endif()
 add_subdirectory(common)
 add_subdirectory(compress)
 add_subdirectory(cwrapper)
+add_subdirectory(dataset)
 add_subdirectory(encoding)
 add_subdirectory(file)
 add_subdirectory(reader)
@@ -82,6 +83,7 @@ set(_TSFILE_OBJECT_TARGETS
         common_obj
         compress_obj
         cwrapper_obj
+        dataset_obj
         file_obj
         read_obj
         write_obj)

cpp/src/common/tsfile_common.h

@@ -351,6 +351,8 @@ class TimeseriesIndex : public ITimeseriesIndex {
     TimeseriesIndex()
         : timeseries_meta_type_((char)255),
           chunk_meta_list_data_size_(0),
+          metadata_offset_(-1),
+          metadata_length_(0),
           measurement_name_(),
           data_type_(common::INVALID_DATATYPE),
           statistic_(nullptr),
@@ -369,6 +371,8 @@ class TimeseriesIndex : public ITimeseriesIndex {
     {
         timeseries_meta_type_ = 0;
         chunk_meta_list_data_size_ = 0;
+        metadata_offset_ = -1;
+        metadata_length_ = 0;
         measurement_name_.reset();
         data_type_ = common::VECTOR;
         chunk_meta_list_serialized_buf_.reset();
@@ -401,6 +405,16 @@ class TimeseriesIndex : public ITimeseriesIndex {
     FORCE_INLINE virtual common::TSDataType get_data_type() const {
         return data_type_;
     }
+    FORCE_INLINE void set_metadata_range(int64_t offset, uint32_t length) {
+        metadata_offset_ = offset;
+        metadata_length_ = length;
+    }
+    FORCE_INLINE int64_t get_metadata_offset() const {
+        return metadata_offset_;
+    }
+    FORCE_INLINE uint32_t get_metadata_length() const {
+        return metadata_length_;
+    }
     int init_statistic(common::TSDataType data_type) {
         if (statistic_ != nullptr &&
             !statistic_from_pa_) {  // clear old statistic
@@ -488,6 +502,8 @@ class TimeseriesIndex : public ITimeseriesIndex {
         int ret = common::E_OK;
         timeseries_meta_type_ = that.timeseries_meta_type_;
         chunk_meta_list_data_size_ = that.chunk_meta_list_data_size_;
+        metadata_offset_ = that.metadata_offset_;
+        metadata_length_ = that.metadata_length_;
         data_type_ = that.data_type_;
 
         statistic_ = StatisticFactory::alloc_statistic_with_pa(data_type_, pa);
@@ -560,6 +576,12 @@ class TimeseriesIndex : public ITimeseriesIndex {
     // Sum of chunk meta serialized size in List<ChunkMeta> of this timeseries.
     uint32_t chunk_meta_list_data_size_;
 
+    // Exact byte range of this TimeseriesMetadata in the source TsFile.
+    // It is assigned by TsFileIOReader after deserialization and is not part
+    // of the on-wire TimeseriesMetadata encoding.
+    int64_t metadata_offset_;
+    uint32_t metadata_length_;
+
     // std::string measurement_name_;
     common::String measurement_name_;
     common::TSDataType data_type_;

cpp/src/cwrapper/tsfile_cwrapper.cc

@@ -419,6 +419,138 @@ ERRNO tsfile_writer_write(TsFileWriter writer, Tablet tablet) {
 
 // Query
 
+PreparedSeriesHandle tsfile_reader_prepare_series(
+    TsFileReader reader, const TsFilePreparedLocator* locator,
+    ERRNO* err_code) {
+    if (err_code == nullptr) {
+        return nullptr;
+    }
+    *err_code = common::E_INVALID_ARG;
+    if (reader == nullptr || locator == nullptr) {
+        return nullptr;
+    }
+    storage::FileGeneration generation;
+    generation.mapped_index_identity = locator->mapped_index_identity;
+    generation.file_id = locator->file_id;
+    generation.file_size = locator->file_size;
+    generation.file_fingerprint = locator->file_fingerprint;
+    storage::PreparedLocator native_locator;
+    native_locator.locator_id = locator->locator_id;
+    native_locator.layout = locator->layout;
+    native_locator.flags = locator->flags;
+    native_locator.value_metadata_offset = locator->value_metadata_offset;
+    native_locator.value_metadata_length = locator->value_metadata_length;
+    native_locator.time_metadata_offset = locator->time_metadata_offset;
+    native_locator.time_metadata_length = locator->time_metadata_length;
+    std::shared_ptr<storage::PreparedSeries> prepared;
+    *err_code = static_cast<storage::TsFileReader*>(reader)->prepare_series(
+        generation, native_locator, prepared);
+    if (*err_code != common::E_OK) {
+        return nullptr;
+    }
+    auto* handle = new (std::nothrow)
+        std::shared_ptr<storage::PreparedSeries>(std::move(prepared));
+    if (handle == nullptr) {
+        *err_code = common::E_OOM;
+    }
+    return handle;
+}
+
+PreparedSeriesHandle tsfile_reader_prepare_series_with_time_owner(
+    TsFileReader reader, const TsFilePreparedLocator* locator,
+    PreparedSeriesHandle aligned_time_owner, ERRNO* err_code) {
+    if (err_code == nullptr) {
+        return nullptr;
+    }
+    *err_code = common::E_INVALID_ARG;
+    if (reader == nullptr || locator == nullptr ||
+        aligned_time_owner == nullptr) {
+        return nullptr;
+    }
+
+    storage::FileGeneration generation;
+    generation.mapped_index_identity = locator->mapped_index_identity;
+    generation.file_id = locator->file_id;
+    generation.file_size = locator->file_size;
+    generation.file_fingerprint = locator->file_fingerprint;
+    storage::PreparedLocator native_locator;
+    native_locator.locator_id = locator->locator_id;
+    native_locator.layout = locator->layout;
+    native_locator.flags = locator->flags;
+    native_locator.value_metadata_offset = locator->value_metadata_offset;
+    native_locator.value_metadata_length = locator->value_metadata_length;
+    native_locator.time_metadata_offset = locator->time_metadata_offset;
+    native_locator.time_metadata_length = locator->time_metadata_length;
+
+    auto* owner = static_cast<std::shared_ptr<storage::PreparedSeries>*>(
+        aligned_time_owner);
+    std::shared_ptr<storage::PreparedSeries> prepared;
+    *err_code = static_cast<storage::TsFileReader*>(reader)->prepare_series(
+        generation, native_locator, *owner, prepared);
+    if (*err_code != common::E_OK) {
+        return nullptr;
+    }
+    auto* handle = new (std::nothrow)
+        std::shared_ptr<storage::PreparedSeries>(std::move(prepared));
+    if (handle == nullptr) {
+        *err_code = common::E_OOM;
+    }
+    return handle;
+}
+
+void tsfile_prepared_series_free(PreparedSeriesHandle prepared) {
+    delete static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared);
+}
+
+ResultSet tsfile_reader_query_prepared(TsFileReader reader,
+                                       PreparedSeriesHandle prepared,
+                                       Timestamp start_time, Timestamp end_time,
+                                       int offset, int limit, ERRNO* err_code) {
+    if (err_code == nullptr) {
+        return nullptr;
+    }
+    *err_code = common::E_INVALID_ARG;
+    if (reader == nullptr || prepared == nullptr) {
+        return nullptr;
+    }
+    auto* handle =
+        static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared);
+    storage::ResultSet* result = nullptr;
+    *err_code = static_cast<storage::TsFileReader*>(reader)->query_prepared(
+        *handle, start_time, end_time, offset, limit, result);
+    return result;
+}
+
+ResultSet tsfile_reader_query_prepared_multi(
+    TsFileReader reader, const PreparedSeriesHandle* prepared,
+    uint32_t prepared_count, Timestamp start_time, Timestamp end_time,
+    int offset, int limit, ERRNO* err_code) {
+    if (err_code == nullptr) {
+        return nullptr;
+    }
+    *err_code = common::E_INVALID_ARG;
+    if (reader == nullptr || prepared == nullptr || prepared_count == 0) {
+        return nullptr;
+    }
+
+    std::vector<std::shared_ptr<storage::PreparedSeries>> native_prepared;
+    native_prepared.reserve(prepared_count);
+    for (uint32_t i = 0; i < prepared_count; i++) {
+        if (prepared[i] == nullptr) {
+            return nullptr;
+        }
+        auto* handle =
+            static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared[i]);
+        native_prepared.push_back(*handle);
+    }
+
+    storage::ResultSet* result = nullptr;
+    *err_code =
+        static_cast<storage::TsFileReader*>(reader)->query_prepared_multi(
+            native_prepared, start_time, end_time, offset, limit, result);
+    return result;
+}
+
 ResultSet tsfile_query_table(TsFileReader reader, const char* table_name,
                              char** columns, uint32_t column_num,
                              Timestamp start_time, Timestamp end_time,
@@ -1305,8 +1437,32 @@ ERRNO populate_c_metadata_map_from_cpp(
                 aligned_idx->value_ts_idx_ != nullptr) {
                 m.data_type = static_cast<TSDataType>(
                     aligned_idx->value_ts_idx_->get_data_type());
+                const storage::TimeseriesIndex* value_idx =
+                    aligned_idx->value_ts_idx_;
+                const storage::TimeseriesIndex* time_idx =
+                    aligned_idx->time_ts_idx_;
+                if (value_idx->get_metadata_offset() >= 0) {
+                    m.value_metadata_offset =
+                        static_cast<uint64_t>(value_idx->get_metadata_offset());
+                    m.value_metadata_length = value_idx->get_metadata_length();
+                }
+                if (time_idx != nullptr &&
+                    time_idx->get_metadata_offset() >= 0) {
+                    m.time_metadata_offset =
+                        static_cast<uint64_t>(time_idx->get_metadata_offset());
+                    m.time_metadata_length = time_idx->get_metadata_length();
+                }
+                m.layout = 1;
             } else {
                 m.data_type = static_cast<TSDataType>(idx->get_data_type());
+                const storage::TimeseriesIndex* value_idx =
+                    dynamic_cast<const storage::TimeseriesIndex*>(idx.get());
+                if (value_idx != nullptr &&
+                    value_idx->get_metadata_offset() >= 0) {
+                    m.value_metadata_offset =
+                        static_cast<uint64_t>(value_idx->get_metadata_offset());
+                    m.value_metadata_length = value_idx->get_metadata_length();
+                }
             }
             storage::Statistic* st = idx->get_statistic();
             int32_t chunk_cnt = 0;
@@ -1316,6 +1472,21 @@ ERRNO populate_c_metadata_map_from_cpp(
                 chunk_cnt = static_cast<int32_t>(cl->size());
             }
             m.chunk_meta_count = chunk_cnt;
+            if (chunk_cnt >= 0 && m.value_metadata_length > 0) {
+                m.locator_flags |= 1;
+            }
+            if (aligned_idx != nullptr) {
+                auto* time_chunks = idx->get_time_chunk_meta_list();
+                if (time_chunks != nullptr) {
+                    m.time_chunk_meta_count =
+                        static_cast<uint32_t>(time_chunks->size());
+                }
+                if (m.time_metadata_length == 0 ||
+                    m.time_chunk_meta_count !=
+                        static_cast<uint32_t>(chunk_cnt)) {
+                    m.locator_flags &= ~static_cast<uint16_t>(1);
+                }
+            }
             const int st_rc = fill_timeseries_statistic(st, &m.statistic);
             if (st_rc != common::E_OK) {
                 for (uint32_t u = 0; u < slot; u++) {