Gabriel39 commented on code in PR #70216:
URL: https://github.com/apache/doris/pull/70216#discussion_r4488702155
##########
be/src/format_v2/table/fluss_union_lake_reader.cpp:
##########
@@ -65,6 +71,185 @@ constexpr const char* PROP_LOG_STOP_OFFSET =
"fluss.log_stop_offset";
constexpr size_t TAIL_BATCH_ROWS = 4096;
+// Marks the rows of one block of lake keys that the tail's sorted keys name,
by walking both in
+// ascending order: a comparison or two per row instead of a hash lookup. A
primary-key table's lake files are
+// sorted by key, so a block read from one ascends; one that does not is left
unmarked and false is returned, so
+// the caller falls back to its hash set. A lake that jumps ahead (a filter, a
sparse file) is caught up by
+// galloping rather than by stepping through every key in between.
+template <typename T>
+bool union_walk_sorted_keys(const std::vector<int64_t>& sorted, const T* data,
size_t rows,
+ UInt8* keep, int64_t* dropped) {
+ if (rows == 0) {
+ return true;
+ }
+ if (!std::is_sorted(data, data + rows)) {
+ return false;
+ }
+ const auto begin = sorted.begin();
+ const size_t size = sorted.size();
+ size_t at = std::lower_bound(begin, sorted.end(),
static_cast<int64_t>(data[0])) - begin;
+ for (size_t row = 0; row < rows; ++row) {
+ const auto value = static_cast<int64_t>(data[row]);
+ if (at < size && sorted[at] < value) {
+ size_t step = 1;
+ while (at + step < size && sorted[at + step] < value) {
+ at += step;
+ step <<= 1;
+ }
+ at = std::lower_bound(begin + at, begin + std::min(size, at + step
+ 1), value) - begin;
+ }
+ const bool replaced = at < size && sorted[at] == value;
+ keep[row] = !replaced;
+ *dropped += replaced;
+ }
+ return true;
+}
+
+// The class a key column packs as, or TYPE_NULL when it cannot be packed. A
packed value is
+// the bytes the value is stored as, so a type qualifies only when two of its
values are equal exactly
+// when those bytes are: not floating point (0.0 and -0.0 compare equal), not
the legacy date and
+// decimal types. The string types pack alike, as the generic predicate
compares them as bytes.
+PrimitiveType union_packing_type(PrimitiveType type) {
+ switch (type) {
+ case TYPE_BOOLEAN:
+ case TYPE_TINYINT:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ case TYPE_LARGEINT:
+ case TYPE_DATEV2:
+ case TYPE_DATETIMEV2:
+ case TYPE_TIMESTAMPTZ:
+ case TYPE_DECIMAL32:
+ case TYPE_DECIMAL64:
+ case TYPE_DECIMAL128I:
+ case TYPE_DECIMAL256:
+ return type;
+ case TYPE_CHAR:
+ case TYPE_VARCHAR:
+ case TYPE_STRING:
+ case TYPE_VARBINARY:
+ return TYPE_STRING;
+ default:
+ return TYPE_NULL;
+ }
+}
+
+// One key column as packing reads it.
+struct UnionPackingColumn {
+ // Keeps a constant column, expanded to one value per row, alive while the
block is probed.
+ ColumnPtr holder;
+ const IColumn* data = nullptr;
+ const UInt8* nulls = nullptr;
+};
+
+// The key columns at `positions`, with the class each packs as; false when
one cannot be packed.
+bool union_packing_columns(const Block& block, const std::vector<size_t>&
positions,
+ std::vector<PrimitiveType>* types,
+ std::vector<UnionPackingColumn>* columns) {
+ types->clear();
+ columns->clear();
+ for (const auto position : positions) {
+ const auto& entry = block.get_by_position(position);
+ const auto type =
union_packing_type(remove_nullable(entry.type)->get_primitive_type());
+ if (type == TYPE_NULL) {
+ return false;
+ }
+ UnionPackingColumn column;
+ column.holder = entry.column->convert_to_full_column_if_const();
+ column.data = column.holder.get();
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(*column.data)) {
+ column.nulls = nullable->get_null_map_data().data();
+ column.data = &nullable->get_nested_column();
+ }
+ types->push_back(type);
+ columns->push_back(std::move(column));
+ }
+ return true;
+}
+
+// One key appended as bytes. Per column a null byte, then the value's length
and bytes; the
+// length keeps ("ab", "c") and ("a", "bc") apart.
+void union_pack_key(const std::vector<UnionPackingColumn>& columns, size_t
row, std::string* out) {
+ for (const auto& column : columns) {
+ if (column.nulls != nullptr && column.nulls[row] != 0) {
+ out->push_back('\1');
+ continue;
+ }
+ out->push_back('\0');
+ const auto value = column.data->get_data_at(row);
+ const auto size = cast_set<uint32_t>(value.size);
+ out->append(reinterpret_cast<const char*>(&size), sizeof(size));
+ out->append(value.data, value.size);
+ }
+}
+
+// Immutable, versioned tails may outlive a query. Use the normal byte-bounded
BE cache so
+// retained columns are charged to the cache tracker and memory GC can reclaim
idle entries.
+class ProcessTailCache final : public LRUCachePolicy {
+public:
+ using Keys = FlussUnionLakeReader::SuppressionKeys;
+
+ static ProcessTailCache& instance() {
+ static ProcessTailCache cache;
Review Comment:
Fixed in 0f4434dcf83. ExecEnv now initializes the tail cache after
CacheManager and releases it after scanner/query teardown, before the manager
and memory infrastructure. The new reader test verifies that cached keys are
released during that teardown and fails with the former static-cache behavior.
An additional local ASAN probe compiles CachePolicy without BE_TEST and
verifies actual unregistration before clearing the manager pointer. All 39
reader tests and the probe pass; the teardown/collision checks also passed 20
repeated runs.
##########
be/src/format_v2/table/fluss_union_lake_reader.cpp:
##########
@@ -65,6 +71,185 @@ constexpr const char* PROP_LOG_STOP_OFFSET =
"fluss.log_stop_offset";
constexpr size_t TAIL_BATCH_ROWS = 4096;
+// Marks the rows of one block of lake keys that the tail's sorted keys name,
by walking both in
+// ascending order: a comparison or two per row instead of a hash lookup. A
primary-key table's lake files are
+// sorted by key, so a block read from one ascends; one that does not is left
unmarked and false is returned, so
+// the caller falls back to its hash set. A lake that jumps ahead (a filter, a
sparse file) is caught up by
+// galloping rather than by stepping through every key in between.
+template <typename T>
+bool union_walk_sorted_keys(const std::vector<int64_t>& sorted, const T* data,
size_t rows,
+ UInt8* keep, int64_t* dropped) {
+ if (rows == 0) {
+ return true;
+ }
+ if (!std::is_sorted(data, data + rows)) {
+ return false;
+ }
+ const auto begin = sorted.begin();
+ const size_t size = sorted.size();
+ size_t at = std::lower_bound(begin, sorted.end(),
static_cast<int64_t>(data[0])) - begin;
+ for (size_t row = 0; row < rows; ++row) {
+ const auto value = static_cast<int64_t>(data[row]);
+ if (at < size && sorted[at] < value) {
+ size_t step = 1;
+ while (at + step < size && sorted[at + step] < value) {
+ at += step;
+ step <<= 1;
+ }
+ at = std::lower_bound(begin + at, begin + std::min(size, at + step
+ 1), value) - begin;
+ }
+ const bool replaced = at < size && sorted[at] == value;
+ keep[row] = !replaced;
+ *dropped += replaced;
+ }
+ return true;
+}
+
+// The class a key column packs as, or TYPE_NULL when it cannot be packed. A
packed value is
+// the bytes the value is stored as, so a type qualifies only when two of its
values are equal exactly
+// when those bytes are: not floating point (0.0 and -0.0 compare equal), not
the legacy date and
+// decimal types. The string types pack alike, as the generic predicate
compares them as bytes.
+PrimitiveType union_packing_type(PrimitiveType type) {
+ switch (type) {
+ case TYPE_BOOLEAN:
+ case TYPE_TINYINT:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ case TYPE_LARGEINT:
+ case TYPE_DATEV2:
+ case TYPE_DATETIMEV2:
+ case TYPE_TIMESTAMPTZ:
+ case TYPE_DECIMAL32:
+ case TYPE_DECIMAL64:
+ case TYPE_DECIMAL128I:
+ case TYPE_DECIMAL256:
+ return type;
+ case TYPE_CHAR:
+ case TYPE_VARCHAR:
+ case TYPE_STRING:
+ case TYPE_VARBINARY:
+ return TYPE_STRING;
+ default:
+ return TYPE_NULL;
+ }
+}
+
+// One key column as packing reads it.
+struct UnionPackingColumn {
+ // Keeps a constant column, expanded to one value per row, alive while the
block is probed.
+ ColumnPtr holder;
+ const IColumn* data = nullptr;
+ const UInt8* nulls = nullptr;
+};
+
+// The key columns at `positions`, with the class each packs as; false when
one cannot be packed.
+bool union_packing_columns(const Block& block, const std::vector<size_t>&
positions,
+ std::vector<PrimitiveType>* types,
+ std::vector<UnionPackingColumn>* columns) {
+ types->clear();
+ columns->clear();
+ for (const auto position : positions) {
+ const auto& entry = block.get_by_position(position);
+ const auto type =
union_packing_type(remove_nullable(entry.type)->get_primitive_type());
+ if (type == TYPE_NULL) {
+ return false;
+ }
+ UnionPackingColumn column;
+ column.holder = entry.column->convert_to_full_column_if_const();
+ column.data = column.holder.get();
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(*column.data)) {
+ column.nulls = nullable->get_null_map_data().data();
+ column.data = &nullable->get_nested_column();
+ }
+ types->push_back(type);
+ columns->push_back(std::move(column));
+ }
+ return true;
+}
+
+// One key appended as bytes. Per column a null byte, then the value's length
and bytes; the
+// length keeps ("ab", "c") and ("a", "bc") apart.
+void union_pack_key(const std::vector<UnionPackingColumn>& columns, size_t
row, std::string* out) {
+ for (const auto& column : columns) {
+ if (column.nulls != nullptr && column.nulls[row] != 0) {
+ out->push_back('\1');
+ continue;
+ }
+ out->push_back('\0');
+ const auto value = column.data->get_data_at(row);
+ const auto size = cast_set<uint32_t>(value.size);
+ out->append(reinterpret_cast<const char*>(&size), sizeof(size));
+ out->append(value.data, value.size);
+ }
+}
+
+// Immutable, versioned tails may outlive a query. Use the normal byte-bounded
BE cache so
+// retained columns are charged to the cache tracker and memory GC can reclaim
idle entries.
+class ProcessTailCache final : public LRUCachePolicy {
+public:
+ using Keys = FlussUnionLakeReader::SuppressionKeys;
+
+ static ProcessTailCache& instance() {
+ static ProcessTailCache cache;
+ return cache;
+ }
+
+ Status get(const std::string& key, const std::function<Status(Keys*)>&
load,
+ std::shared_ptr<const Keys>* result, bool* hit) {
+ // A stripe serializes concurrent misses without holding the LRU lock
across network I/O.
+ std::lock_guard<std::mutex> guard(
+ _loading[std::hash<std::string>()(key) % _loading.size()]);
Review Comment:
Fixed in 0f4434dcf83. The cache now looks up and pins hits before acquiring
a load stripe. Misses recheck under the stripe so concurrent loads still
coalesce. The regression test chooses two colliding tail keys, pauses the cold
JNI load, and verifies that the cached tail completes and returns the correct
suppressed rows before that load is released. It fails with the previous lock
ordering and passes with the fix, including 20 repeated runs.
##########
fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussConnector.java:
##########
@@ -94,10 +94,26 @@ public class FlussConnector implements Connector {
// database and table, and targeted invalidation must follow the same path
as lookup.
private final Map<TablePath, TablePath> lakeTablePaths = new
LinkedHashMap<>();
+ // Outlives the per-statement admin ops; see FlussTableInfoCache.
+ private final FlussTableInfoCache tableInfos;
+
+ // Outlives the per-node scan providers; see FlussLakePropertiesCache.
+ private final FlussLakePropertiesCache lakeProperties;
+
+ // Each metadata scope owns its pruning list so a new statement cannot
miss newly created partitions.
+ private final long partitionCacheTtlSec;
+
public FlussConnector(FlussCatalogProperties properties, ConnectorContext
context) {
this.properties = properties;
this.catalogName = context.getCatalogName();
this.context = context;
+ this.tableInfos = new
FlussTableInfoCache(Long.parseLong(properties.getRawCatalogProperties()
Review Comment:
Fixed in 0f4434dcf83. Both TTLs are now bound by FlussCatalogProperties, so
malformed and overflowing values are rejected by CREATE/merged ALTER validation
before persistence; connector construction uses the validated fields. Tests
cover both keys, ALTER validation without mutating current properties,
whitespace, defaults, disable values and long boundaries. The validated
metadata TTL is also explicitly forwarded to JNI under its Doris key, while
both TTLs stay out of the Fluss client configuration. All 302 connector tests
and 97 scanner tests pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
