Merge branch 'apache:master' into s7

Author: Tech-Circle-48

.github/workflows/build-extension.yml

@@ -40,6 +40,7 @@ jobs:
     outputs:
       broker_changes: ${{ steps.filter.outputs.broker_changes }}
       docs_changes: ${{ steps.filter.outputs.docs_changes }}
+      cdc_client_changes: ${{ steps.filter.outputs.cdc_client_changes }}
     steps:
       - name: Checkout ${{ github.ref }}
         uses: actions/checkout@v3
@@ -53,9 +54,11 @@ jobs:
         with:
           filters: |
             broker_changes:
-              - 'fs_brokers/**'
+              - 'fs_brokers/apache_hdfs_broker/**'
             docs_changes:
               - 'docs/**'
+            cdc_client_changes:
+              - 'fs_brokers/cdc_client/**'
   build-broker:
     name: Build Broker
     needs: changes
@@ -92,6 +95,41 @@ jobs:
       - name: Build broker
         run: |
           cd fs_brokers/apache_hdfs_broker/ && /bin/bash build.sh
+  build-cdc-client:
+    name: Build Cdc Client
+    needs: changes
+    if: ${{ needs.changes.outputs.cdc_client_changes == 'true' }}
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout ${{ github.ref }}
+        uses: actions/checkout@v3
+
+      - name: Setup java
+        uses: actions/setup-java@v2
+        with:
+          distribution: adopt
+          java-version: '17'
+
+      - name: Setup thrift
+        run: |
+          pushd thirdparty
+          branch="${{ github.base_ref }}"
+          if [[ -z "${branch}" ]] || [[ "${branch}" == 'master' || "${branch}" == 'branch-4.0' || "${branch}" == 'branch-3.0' || "${branch}" == 'branch-2.1' ]]; then
+            curl -L https://github.com/apache/doris-thirdparty/releases/download/automation/doris-thirdparty-prebuilt-linux-x86_64.tar.xz \
+              -o doris-thirdparty-prebuilt-linux-x86_64.tar.xz
+          else
+            curl -L "https://github.com/apache/doris-thirdparty/releases/download/automation-${branch/branch-/}/doris-thirdparty-prebuilt-linux-x86_64.tar.xz" \
+              -o doris-thirdparty-prebuilt-linux-x86_64.tar.xz
+          fi
+          tar -xvf doris-thirdparty-prebuilt-linux-x86_64.tar.xz
+          popd
+          export PATH="${DEFAULT_DIR}/ldb-toolchain/bin/:$(pwd)/thirdparty/installed/bin/:${PATH}"
+          
+          thrift --version
+
+      - name: Build cdc client
+        run: |
+          cd fs_brokers/cdc_client/ && /bin/bash build.sh
   # build-docs:
   #   name: Build Documents
   #   needs: changes

be/src/agent/task_worker_pool.cpp

@@ -2399,7 +2399,7 @@ void clean_udf_cache_callback(const TAgentTaskRequest& req) {
     if (doris::config::enable_java_support) {
         LOG(INFO) << "clean udf cache start: " << req.clean_udf_cache_req.function_signature;
         static_cast<void>(
-                JniUtil::clean_udf_class_load_cache(req.clean_udf_cache_req.function_signature));
+                Jni::Util::clean_udf_class_load_cache(req.clean_udf_cache_req.function_signature));
         LOG(INFO) << "clean udf cache  finish: " << req.clean_udf_cache_req.function_signature;
     }
 }

be/src/cloud/cloud_rowset_writer.cpp

@@ -195,7 +195,7 @@ Status CloudRowsetWriter::_collect_packed_slice_location(io::FileWriter* file_wr
     }
 
     rowset_meta->add_packed_slice_location(file_path, index.packed_file_path, index.offset,
-                                           index.size);
+                                           index.size, index.packed_file_size);
     LOG(INFO) << "collect packed file index: " << file_path << " -> " << index.packed_file_path
               << ", offset: " << index.offset << ", size: " << index.size;
     return Status::OK();

be/src/exec/olap_utils.h

@@ -104,23 +104,23 @@ inline SQLFilterOp to_olap_filter_type(TExprOpcode::type type, bool opposite) {
     return FILTER_IN;
 }
 
-inline SQLFilterOp to_olap_filter_type(const std::string& function_name, bool opposite) {
+inline SQLFilterOp to_olap_filter_type(const std::string& function_name) {
     if (function_name == "lt") {
-        return opposite ? FILTER_LARGER : FILTER_LESS;
+        return FILTER_LESS;
     } else if (function_name == "gt") {
-        return opposite ? FILTER_LESS : FILTER_LARGER;
+        return FILTER_LARGER;
     } else if (function_name == "le") {
-        return opposite ? FILTER_LARGER_OR_EQUAL : FILTER_LESS_OR_EQUAL;
+        return FILTER_LESS_OR_EQUAL;
     } else if (function_name == "ge") {
-        return opposite ? FILTER_LESS_OR_EQUAL : FILTER_LARGER_OR_EQUAL;
+        return FILTER_LARGER_OR_EQUAL;
     } else if (function_name == "eq") {
-        return opposite ? FILTER_NOT_IN : FILTER_IN;
+        return FILTER_IN;
     } else if (function_name == "ne") {
-        return opposite ? FILTER_IN : FILTER_NOT_IN;
+        return FILTER_NOT_IN;
     } else if (function_name == "in") {
-        return opposite ? FILTER_NOT_IN : FILTER_IN;
+        return FILTER_IN;
     } else if (function_name == "not_in") {
-        return opposite ? FILTER_IN : FILTER_NOT_IN;
+        return FILTER_NOT_IN;
     } else {
         DCHECK(false) << "Function Name: " << function_name;
         return FILTER_IN;

be/src/exprs/create_predicate_function.h

@@ -268,11 +268,13 @@ std::shared_ptr<ColumnPredicate> create_olap_column_predicate(
         const TabletColumn* column, bool) {
     // currently only support like predicate
     if constexpr (PT == TYPE_CHAR) {
-        return LikeColumnPredicate<TYPE_CHAR>::create_shared(
-                filter->_opposite, column_id, filter->_fn_ctx, filter->_string_param);
+        return LikeColumnPredicate<TYPE_CHAR>::create_shared(filter->_opposite, column_id,
+                                                             column->name(), filter->_fn_ctx,
+                                                             filter->_string_param);
     } else if constexpr (PT == TYPE_VARCHAR || PT == TYPE_STRING) {
-        return LikeColumnPredicate<TYPE_STRING>::create_shared(
-                filter->_opposite, column_id, filter->_fn_ctx, filter->_string_param);
+        return LikeColumnPredicate<TYPE_STRING>::create_shared(filter->_opposite, column_id,
+                                                               column->name(), filter->_fn_ctx,
+                                                               filter->_string_param);
     }
     throw Exception(ErrorCode::INTERNAL_ERROR, "function filter do not support type {}", PT);
 }

be/src/io/fs/benchmark/benchmark_factory.hpp

@@ -104,7 +104,7 @@ class MultiBenchmark {
         Status status = Status::OK();
         if (doris::config::enable_java_support) {
             // Init jni
-            status = doris::JniUtil::Init();
+            status = doris::Jni::Util::Init();
             if (!status.ok()) {
                 LOG(WARNING) << "Failed to initialize JNI: " << status;
                 exit(1);

be/src/io/fs/hdfs_file_writer.cpp

@@ -128,7 +128,7 @@ class HdfsWriteMemUsageRecorder {
 private:
     // clang-format off
     size_t max_jvm_heap_size() const {
-        return JniUtil::get_max_jni_heap_memory_size();
+        return Jni::Util::get_max_jni_heap_memory_size();
     }
     // clang-format on
     [[maybe_unused]] std::size_t cur_memory_comsuption {0};

be/src/io/fs/packed_file_manager.cpp

@@ -150,6 +150,8 @@ Status PackedFileManager::create_new_packed_file_context(
     // Create file writer for the packed file
     FileWriterPtr new_writer;
     FileWriterOptions opts;
+    // enable write file cache for packed file
+    opts.write_file_cache = true;
     RETURN_IF_ERROR(
             packed_file_ctx->file_system->create_file(Path(relative_path), &new_writer, &opts));
     packed_file_ctx->writer = std::move(new_writer);
@@ -256,6 +258,7 @@ Status PackedFileManager::append_small_file(const std::string& path, const Slice
     location.packed_file_path = active_state->packed_file_path;
     location.offset = active_state->current_offset;
     location.size = data.get_size();
+    location.create_time = std::time(nullptr);
     location.tablet_id = info.tablet_id;
     location.rowset_id = info.rowset_id;
     location.resource_id = info.resource_id;
@@ -609,6 +612,15 @@ void PackedFileManager::process_uploading_packed_files() {
             first_slice = false;
             slices_stream << small_file_path << "(txn=" << index.txn_id
                           << ", offset=" << index.offset << ", size=" << index.size << ")";
+
+            // Update packed_file_size in global index
+            {
+                std::lock_guard<std::mutex> global_lock(_global_index_mutex);
+                auto it = _global_slice_locations.find(small_file_path);
+                if (it != _global_slice_locations.end()) {
+                    it->second.packed_file_size = packed_file->total_size;
+                }
+            }
         }
         LOG(INFO) << "Packed file " << packed_file->packed_file_path
                   << " uploaded; slices=" << packed_file->slice_locations.size()
@@ -809,30 +821,12 @@ void PackedFileManager::cleanup_expired_data() {
 
     // Clean up expired global index entries
     {
-        std::unordered_set<std::string> active_packed_files;
-        {
-            std::lock_guard<std::timed_mutex> current_lock(_current_packed_file_mutex);
-            for (const auto& [resource_id, state] : _current_packed_files) {
-                if (state) {
-                    active_packed_files.insert(state->packed_file_path);
-                }
-            }
-        }
-        {
-            std::lock_guard<std::mutex> merge_lock(_packed_files_mutex);
-            for (const auto& [path, state] : _uploading_packed_files) {
-                active_packed_files.insert(path);
-            }
-            for (const auto& [path, state] : _uploaded_packed_files) {
-                active_packed_files.insert(path);
-            }
-        }
-
         std::lock_guard<std::mutex> global_lock(_global_index_mutex);
         auto it = _global_slice_locations.begin();
         while (it != _global_slice_locations.end()) {
             const auto& index = it->second;
-            if (active_packed_files.find(index.packed_file_path) == active_packed_files.end()) {
+            if (index.create_time > 0 &&
+                current_time - index.create_time > config::uploaded_file_retention_seconds) {
                 it = _global_slice_locations.erase(it);
             } else {
                 ++it;

be/src/io/fs/packed_file_manager.h

@@ -45,10 +45,12 @@ struct PackedSliceLocation {
     std::string packed_file_path;
     int64_t offset;
     int64_t size;
+    int64_t create_time = 0;
     int64_t tablet_id = 0;
     std::string rowset_id;
     std::string resource_id;
     int64_t txn_id = 0;
+    int64_t packed_file_size = -1; // Total size of the packed file, -1 means not set
 };
 
 struct PackedAppendContext {

be/src/io/fs/packed_file_system.cpp

@@ -71,13 +71,11 @@ Status PackedFileSystem::open_file_impl(const Path& file, FileReaderSPtr* reader
         FileReaderSPtr inner_reader;
         // Create a new FileReaderOptions with the correct file size
         FileReaderOptions local_opts = opts ? *opts : FileReaderOptions();
-        // DCHECK(opts->file_size == -1 || opts->file_size == index.size)
-        //         << "file size is not correct, expected: " << index.size
-        //         << ", actual: " << opts->file_size;
-        // local_opts.file_size = index.size + index.offset;
-        local_opts.file_size = -1;
+        // Set file_size to packed file size to avoid head object request
+        local_opts.file_size = index.packed_file_size;
         LOG(INFO) << "open packed file: " << index.packed_file_path << ", file: " << file.native()
-                  << ", offset: " << index.offset << ", size: " << index.size;
+                  << ", offset: " << index.offset << ", size: " << index.size
+                  << ", packed_file_size: " << index.packed_file_size;
         RETURN_IF_ERROR(
                 _inner_fs->open_file(Path(index.packed_file_path), &inner_reader, &local_opts));