Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions benchmark/micro/variant/variant_unshredded_path_extract.benchmark
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
# name: benchmark/micro/variant/variant_unshredded_path_extract.benchmark
# description: Filter on a small top-level key whose values live in the unshredded Parquet VARIANT value blob (forced via mismatched shredding types so typed_value cannot hold them). Isolates CPU/allocation cost of pushed path extraction versus converting the full root blob; TOTAL_BYTES_READ is expected to stay similar because the root value page must still be read.
# group: [variant]

name variant_unshredded_path_extract
group variant
subgroup parquet
storage persistent v2.0.0

cache variant_unshredded_path_extract.duckdb

load
COPY (
SELECT {
'id': i::INTEGER,
'flag': (i % 2 = 0),
'payload_a': md5(i::VARCHAR) || md5((i + 1)::VARCHAR) || md5((i + 2)::VARCHAR) || md5((i + 3)::VARCHAR),
'payload_b': md5((i + 10)::VARCHAR) || md5((i + 11)::VARCHAR) || md5((i + 12)::VARCHAR) || md5((i + 13)::VARCHAR),
'payload_c': md5((i + 20)::VARCHAR) || md5((i + 21)::VARCHAR) || md5((i + 22)::VARCHAR) || md5((i + 23)::VARCHAR),
'nested': {
'x': (i % 100)::INTEGER,
'y': md5((i + 30)::VARCHAR) || md5((i + 31)::VARCHAR)
}
}::VARIANT AS v
FROM range(200000) t(i)
) TO 'variant_unshredded_path_extract.parquet' (
FORMAT PARQUET,
SHREDDING {'v': 'VARCHAR'}
);

run
SELECT count(*) FROM 'variant_unshredded_path_extract.parquet' WHERE try(v.id::INTEGER) = 12345;

result I
1
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,12 @@ class VariantBinaryDecoder {
public:
static VariantValue Decode(const VariantMetadata &metadata, const_data_ptr_t data, idx_t data_offset,
idx_t data_size);
//! Walk a constant key path without decoding sibling fields. Returns MISSING for absent keys /
//! non-object intermediates. Terminal objects/arrays still decode their complete requested subtree.
static VariantValue DecodePath(const VariantMetadata &metadata, const_data_ptr_t data, idx_t data_offset,
idx_t data_size, const vector<string> &path);
//! Rebuild a sparse object tree from 'path' so post-scan variant_extract still resolves the same keys.
static VariantValue WrapSparsePath(VariantValue &&leaf, const vector<string> &path);

public:
static VariantValue PrimitiveTypeDecode(const VariantValueMetadata &value_metadata, const_data_ptr_t data,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@ class VariantShreddedConversion {
VariantShreddedConversion() = delete;

public:
static vector<VariantValue> Convert(Vector &metadata, Vector &group, idx_t offset, idx_t length, idx_t total_size);
static vector<VariantValue> Convert(Vector &metadata, Vector &group, idx_t offset, idx_t length, idx_t total_size,
const vector<string> *extract_path = nullptr);
static void ConvertBinaryToVariant(Vector &metadata_and_value, idx_t offset, idx_t length, idx_t total_size,
Vector &result);
static vector<VariantValue> ConvertShreddedLeaf(Vector &metadata, Vector &value, Vector &typed_value, idx_t offset,
Expand Down
5 changes: 4 additions & 1 deletion extension/parquet/include/reader/variant_column_reader.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,13 @@ class VariantColumnReader : public ColumnReader {

public:
VariantColumnReader(ClientContext &context, const ParquetReader &reader, const ParquetColumnSchema &schema,
vector<unique_ptr<ColumnReader>> child_readers_p, bool projected_mode_p = false);
vector<unique_ptr<ColumnReader>> child_readers_p, bool projected_mode_p = false,
vector<string> extract_path_p = {});

ClientContext &context;
vector<unique_ptr<ColumnReader>> child_readers;
//! Optional constant key path from extract projection hints (unshredded binary fast path)
vector<string> extract_path;

public:
ColumnReader &GetChildReader(idx_t child_idx);
Expand Down
24 changes: 23 additions & 1 deletion extension/parquet/parquet_reader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -592,10 +592,32 @@ unique_ptr<ColumnReader> ParquetReader::CreateReaderRecursive(ClientContext &con
return make_uniq<VariantColumnReader>(context, *this, schema, std::move(children), true);
}

//! Unshredded / non-projectable path: retain a single constant key path for selective binary decode
vector<string> extract_path;
if (indexes.size() == 1 && !indexes[0].HasPrimaryIndex()) {
reference<const ColumnIndex> current(indexes[0]);
while (true) {
if (current.get().HasPrimaryIndex()) {
extract_path.clear();
break;
}
extract_path.push_back(current.get().GetFieldName());
if (!current.get().HasChildren()) {
break;
}
if (current.get().ChildIndexCount() != 1) {
extract_path.clear();
break;
}
current = current.get().GetChildIndex(0);
}
}

for (idx_t child_index = 0; child_index < schema.children.size(); child_index++) {
children[child_index] = CreateReaderRecursive(context, vector<ColumnIndex>(), schema.children[child_index]);
}
return make_uniq<VariantColumnReader>(context, *this, schema, std::move(children), false);
return make_uniq<VariantColumnReader>(context, *this, schema, std::move(children), false,
std::move(extract_path));
}
default:
throw InternalException("Unsupported ParquetColumnSchemaType");
Expand Down
113 changes: 113 additions & 0 deletions extension/parquet/reader/variant/variant_binary_decoder.cpp
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
#include "reader/variant/variant_binary_decoder.hpp"
#include "duckdb/common/printer.hpp"
#include "duckdb/common/optional_idx.hpp"
#include "duckdb/common/numeric_utils.hpp"
#include "utf8proc_wrapper.hpp"

#include "reader/uuid_column_reader.hpp"
Expand All @@ -10,6 +12,8 @@
#include "duckdb/common/types/time.hpp"
#include "duckdb/common/types/date.hpp"

#include <algorithm>

static constexpr uint8_t VERSION_MASK = 0xF;
static constexpr uint8_t SORTED_STRINGS_MASK = 0x1;
static constexpr uint8_t SORTED_STRINGS_SHIFT = 4;
Expand Down Expand Up @@ -462,4 +466,113 @@ VariantValue VariantBinaryDecoder::Decode(const VariantMetadata &variant_metadat
}
}

namespace {

optional_idx FindMetadataStringId(const VariantMetadata &metadata, const string &key) {
auto &strings = metadata.strings;
//! Prefer binary search when the dictionary claims to be sorted. Some writers set the sorted flag while
//! leaving keys in insertion order (e.g. ["id","b"]), so fall back to a linear scan on miss.
if (metadata.header.sorted_strings) {
auto it = std::lower_bound(strings.begin(), strings.end(), key);
if (it != strings.end() && *it == key) {
return optional_idx(NumericCast<idx_t>(it - strings.begin()));
}
}
for (idx_t i = 0; i < strings.size(); i++) {
if (strings[i] == key) {
return optional_idx(i);
}
}
return optional_idx();
}

//! Locate 'key' in a binary OBJECT and return the byte offset of the matching child value (header byte),
//! or invalid when the key is absent.
optional_idx FindObjectChildOffset(const VariantMetadata &metadata, const VariantValueMetadata &value_metadata,
const_data_ptr_t data, idx_t data_offset, idx_t data_size, const string &key) {
auto field_id = FindMetadataStringId(metadata, key);
if (!field_id.IsValid()) {
return optional_idx();
}

auto field_offset_size = value_metadata.field_offset_size;
auto field_id_size = value_metadata.field_id_size;
auto is_large = value_metadata.is_large;

idx_t num_elements;
idx_t cursor = data_offset;
if (is_large) {
if (cursor + sizeof(uint32_t) > data_size) {
throw IOException("Corrupted VARIANT 'value' buffer");
}
num_elements = Load<uint32_t>(data + cursor);
cursor += sizeof(uint32_t);
} else {
if (cursor + sizeof(uint8_t) > data_size) {
throw IOException("Corrupted VARIANT 'value' buffer");
}
num_elements = Load<uint8_t>(data + cursor);
cursor += sizeof(uint8_t);
}

auto field_ids_offset = cursor;
auto field_offsets_offset = cursor + (num_elements * field_id_size);
auto values_offset = field_offsets_offset + ((num_elements + 1) * field_offset_size);

idx_t ids_cursor = field_ids_offset;
idx_t offsets_cursor = field_offsets_offset;
idx_t last_offset = ReadVariableLengthLittleEndian(field_offset_size, data, offsets_cursor, data_size);
for (idx_t i = 0; i < num_elements; i++) {
auto id = ReadVariableLengthLittleEndian(field_id_size, data, ids_cursor, data_size);
auto next_offset = ReadVariableLengthLittleEndian(field_offset_size, data, offsets_cursor, data_size);
if (id == field_id.GetIndex()) {
auto child_offset = values_offset + last_offset;
if (child_offset >= data_size) {
throw IOException("Corrupted VARIANT 'value' buffer");
}
return optional_idx(child_offset);
}
last_offset = next_offset;
}
return optional_idx();
}

} // namespace

VariantValue VariantBinaryDecoder::DecodePath(const VariantMetadata &metadata, const_data_ptr_t data, idx_t data_offset,
idx_t data_size, const vector<string> &path) {
D_ASSERT(!path.empty());
idx_t offset = data_offset;
for (idx_t path_idx = 0; path_idx < path.size(); path_idx++) {
if (offset + 1 > data_size) {
throw IOException("Corrupted VARIANT 'value' buffer");
}
auto value_metadata = VariantValueMetadata::FromHeaderByte(data[offset]);
if (value_metadata.basic_type != VariantBasicType::OBJECT) {
return VariantValue();
}
auto child_offset =
FindObjectChildOffset(metadata, value_metadata, data, offset + 1, data_size, path[path_idx]);
if (!child_offset.IsValid()) {
return VariantValue();
}
offset = child_offset.GetIndex();
}
//! Decode only the terminal subtree
return Decode(metadata, data, offset, data_size);
}

VariantValue VariantBinaryDecoder::WrapSparsePath(VariantValue &&leaf, const vector<string> &path) {
if (leaf.IsMissing() || path.empty()) {
return std::move(leaf);
}
VariantValue current = std::move(leaf);
for (idx_t i = path.size(); i > 0; i--) {
VariantValue parent(VariantValueType::OBJECT);
parent.AddChild(path[i - 1], std::move(current));
current = std::move(parent);
}
return current;
}

} // namespace duckdb
26 changes: 20 additions & 6 deletions extension/parquet/reader/variant/variant_shredded_conversion.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -324,7 +324,8 @@ struct ShreddedVariantField {
} // namespace

static vector<VariantValue> ConvertBinaryEncoding(Vector &metadata, Vector &value, idx_t offset, idx_t length,
idx_t total_size, bool add_metadata_offset = false) {
idx_t total_size, bool add_metadata_offset = false,
const vector<string> *extract_path = nullptr) {
UnifiedVectorFormat value_format;
value.ToUnifiedFormat(total_size, value_format);
auto value_data = value_format.GetData<string_t>(value_format);
Expand All @@ -334,6 +335,7 @@ static vector<VariantValue> ConvertBinaryEncoding(Vector &metadata, Vector &valu
metadata.ToUnifiedFormat(length, metadata_format);
auto metadata_data = metadata_format.GetData<string_t>(metadata_format);
auto metadata_validity = metadata_format.validity;
(void)metadata_validity;

//! Fills every row with MISSING, turned into NULL later if this is not in an OBJECT field
vector<VariantValue> ret(length);
Expand All @@ -356,8 +358,15 @@ static vector<VariantValue> ConvertBinaryEncoding(Vector &metadata, Vector &valu
//! The value bytes start directly after the metadata bytes
value_offset += variant_metadata.total_size;
}
ret[i] = VariantBinaryDecoder::Decode(variant_metadata, const_data_ptr_cast(binary_value), value_offset,
value_buffer.GetSize());
if (extract_path && !extract_path->empty()) {
//! Selective path decode: skip sibling fields, then wrap so post-scan extract still works
auto leaf = VariantBinaryDecoder::DecodePath(variant_metadata, const_data_ptr_cast(binary_value),
value_offset, value_buffer.GetSize(), *extract_path);
ret[i] = VariantBinaryDecoder::WrapSparsePath(std::move(leaf), *extract_path);
} else {
ret[i] = VariantBinaryDecoder::Decode(variant_metadata, const_data_ptr_cast(binary_value), value_offset,
value_buffer.GetSize());
}
}
return ret;
}
Expand Down Expand Up @@ -532,7 +541,7 @@ vector<VariantValue> VariantShreddedConversion::ConvertShreddedArray(Vector &met
}

vector<VariantValue> VariantShreddedConversion::Convert(Vector &metadata, Vector &group, idx_t offset, idx_t length,
idx_t total_size) {
idx_t total_size, const vector<string> *extract_path) {
D_ASSERT(group.GetType().id() == LogicalTypeId::STRUCT);

auto &group_entries = StructVector::GetEntries(group);
Expand Down Expand Up @@ -560,7 +569,12 @@ vector<VariantValue> VariantShreddedConversion::Convert(Vector &metadata, Vector

if (typed_value) {
auto &type = typed_value->GetType();
vector<VariantValue> ret;
//! When the root is shredded as a leaf (e.g. SHREDDING VARCHAR) the object payload lives in
//! binary 'value'. Prefer the selective path decode over shredding the unusable typed leaf.
if (extract_path && !extract_path->empty() && type.id() != LogicalTypeId::STRUCT &&
type.id() != LogicalTypeId::LIST) {
return ConvertBinaryEncoding(metadata, *value, offset, length, total_size, false, extract_path);
}
if (type.id() == LogicalTypeId::STRUCT) {
return ConvertShreddedObject(metadata, *value, *typed_value, offset, length, total_size);
} else if (type.id() == LogicalTypeId::LIST) {
Expand All @@ -569,7 +583,7 @@ vector<VariantValue> VariantShreddedConversion::Convert(Vector &metadata, Vector
return ConvertShreddedLeaf(metadata, *value, *typed_value, offset, length, total_size);
}
} else {
return ConvertBinaryEncoding(metadata, *value, offset, length, total_size);
return ConvertBinaryEncoding(metadata, *value, offset, length, total_size, false, extract_path);
}
}

Expand Down
10 changes: 6 additions & 4 deletions extension/parquet/reader/variant_column_reader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,10 @@ namespace duckdb {
//===--------------------------------------------------------------------===//
VariantColumnReader::VariantColumnReader(ClientContext &context, const ParquetReader &reader,
const ParquetColumnSchema &schema,
vector<unique_ptr<ColumnReader>> child_readers_p, bool projected_mode_p)
vector<unique_ptr<ColumnReader>> child_readers_p, bool projected_mode_p,
vector<string> extract_path_p)
: ColumnReader(reader, schema), context(context), child_readers(std::move(child_readers_p)),
projected_mode(projected_mode_p) {
extract_path(std::move(extract_path_p)), projected_mode(projected_mode_p) {
D_ASSERT(Type().InternalType() == PhysicalType::STRUCT);

optional_idx metadata_idx;
Expand Down Expand Up @@ -111,8 +112,9 @@ idx_t VariantColumnReader::Read(uint64_t num_values, data_ptr_t define_out, data
"The shredded Variant column did not contain the same amount of values for 'typed_value' and 'value'");
}
}
auto intermediate =
VariantShreddedConversion::Convert(metadata_intermediate, intermediate_group, 0, num_values, num_values);
const vector<string> *path_ptr = extract_path.empty() ? nullptr : &extract_path;
auto intermediate = VariantShreddedConversion::Convert(metadata_intermediate, intermediate_group, 0, num_values,
num_values, path_ptr);
VariantValue::ToVARIANT(intermediate, result);

return value_values;
Expand Down
11 changes: 5 additions & 6 deletions extension/parquet/variant_arrow_capi.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -257,10 +257,10 @@ void RegisterParquetVariantArrowExtension(DatabaseInstance &db_instance) {
if (config.HasArrowExtension(LogicalType::VARIANT())) {
return;
}
config.RegisterArrowExtension({ARROW_PARQUET_VARIANT, PopulateVariantArrowSchema, GetVariantArrowType,
make_shared_ptr<ArrowTypeExtensionData>(LogicalType::VARIANT(),
GetParquetVariantStructType(),
ArrowToVariant, VariantToArrow)});
config.RegisterArrowExtension(
{ARROW_PARQUET_VARIANT, PopulateVariantArrowSchema, GetVariantArrowType,
make_shared_ptr<ArrowTypeExtensionData>(LogicalType::VARIANT(), GetParquetVariantStructType(), ArrowToVariant,
VariantToArrow)});
}

duckdb_state RegisterParquetVariantArrow(duckdb_database database) {
Expand All @@ -277,8 +277,7 @@ duckdb_state RegisterParquetVariantArrow(duckdb_database database) {
extern "C" DUCKDB_EXTENSION_API duckdb_state duckdb_parquet_variant_bytes_to_json(duckdb_database database,
const uint8_t *metadata,
idx_t metadata_len,
const uint8_t *value,
idx_t value_len,
const uint8_t *value, idx_t value_len,
char **out_json) {
if (!database || !out_json) {
return DuckDBError;
Expand Down
3 changes: 2 additions & 1 deletion src/common/arrow/arrow_appender.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,8 @@ ArrowArray ArrowAppender::Finalize() {

for (idx_t i = 0; i < root_holder->child_data.size(); i++) {
auto &child_data = root_holder->child_data[i];
const LogicalType child_type = child_data->extension_data ? child_data->extension_data->GetInternalType() : types[i];
const LogicalType child_type =
child_data->extension_data ? child_data->extension_data->GetInternalType() : types[i];
root_holder->child_arrays[i] = *ArrowAppender::FinalizeChild(child_type, std::move(root_holder->child_data[i]));
}

Expand Down
4 changes: 2 additions & 2 deletions src/include/duckdb/common/multi_file/multi_file_function.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -492,8 +492,8 @@ class MultiFileFunction : public TableFunction {
// optionally apply sampling pushdown after filter pruning, before any readers open
if (input.sample_options) {
auto &candidate_list = new_list ? *new_list : *bind_data.file_list;
auto sampled_list = bind_data.multi_file_reader->SamplePushdown(context, candidate_list,
bind_data.file_options, *input.sample_options);
auto sampled_list = bind_data.multi_file_reader->SamplePushdown(
context, candidate_list, bind_data.file_options, *input.sample_options);
if (sampled_list) {
new_list = std::move(sampled_list);
}
Expand Down
5 changes: 2 additions & 3 deletions src/optimizer/sampling_pushdown.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,8 @@ unique_ptr<LogicalOperator> SamplingPushdown::Optimize(unique_ptr<LogicalOperato
auto &get = op->children[0]->Cast<LogicalGet>();
const auto &sample_options = *sample_op.sample_options;
const bool has_filters = !get.table_filters.filters.empty() || get.dynamic_filters;
const bool can_push_system_sample =
sample_options.method == SampleMethod::SYSTEM_SAMPLE && !has_filters &&
CanPushSampling(context, get, sample_options);
const bool can_push_system_sample = sample_options.method == SampleMethod::SYSTEM_SAMPLE && !has_filters &&
CanPushSampling(context, get, sample_options);

if (can_push_system_sample) {
const bool is_row_count_sampling = !sample_options.is_percentage;
Expand Down
Loading
Loading