github-actions[bot] commented on code in PR #70216:
URL: https://github.com/apache/doris/pull/70216#discussion_r4488501123
##########
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:
[P2] Destroy the tail cache before the cache manager
After a versioned Fluss union read constructs this static, shutting down
with `enable_graceful_exit_check=true` calls `ExecEnv::destroy()` before
returning from `main`. That deletes `_cache_manager`, but this cache is
destroyed only afterward: `CachePolicy::~CachePolicy()` then calls
`CacheManager::instance()->unregister_cache()` through the cleared manager
pointer and crashes the shutdown. Give the cache an explicit lifecycle that
releases/unregisters it before the cache manager and memory infrastructure. The
new unit test misses this because `unregister_cache` returns early under
`BE_TEST`.
##########
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:
[P2] Serve cache hits before locking the load stripe
The mutex is acquired before `lookup`, so a cold tail A holds its stripe for
the entire JNI log-tail read and also blocks an already-cached, unrelated tail
B with the same `hash % 32`. With up to 256 retained entries, these collisions
can make warm scans wait for another query's full network read while occupying
scanner workers; cancellation cannot interrupt the mutex wait. Look up and pin
existing entries before taking the load stripe, then recheck after locking on a
miss, so coalescing misses does not serialize cache hits behind I/O.
##########
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:
[P2] Validate TTL properties before persisting an ALTER
Both TTLs are parsed only when the connector is constructed, while
CREATE/ALTER validation calls `FlussCatalogProperties.of()`, which does not
validate either key. Consequently `ALTER CATALOG ... SET PROPERTIES
('fluss.metadata.cache.ttl.sec'='invalid')` succeeds and journals the value;
after the runtime reset, subsequent table/metadata reads fail here with
`NumberFormatException` until another ALTER repairs it. The partition TTL and
overflowing numbers have the same problem. Bind/validate these recognized
properties in the holder used by statement-time validation, then consume its
validated values here.
--
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]
