Merge branch 'duckdb:master' into master

Author: TNonet

data/parquet-testing/adam_genotypes.parquet

@@ -174,6 +174,7 @@ LogicalType ParquetReader::DeriveLogicalType(const SchemaElement &s_ele, bool bi
 				    "DECIMAL converted type can only be set for value of Type::(FIXED_LEN_)BYTE_ARRAY/INT32/INT64");
 			}
 		case ConvertedType::UTF8:
+		case ConvertedType::ENUM:
 			switch (s_ele.type) {
 			case Type::BYTE_ARRAY:
 			case Type::FIXED_LEN_BYTE_ARRAY:
@@ -193,7 +194,6 @@ LogicalType ParquetReader::DeriveLogicalType(const SchemaElement &s_ele, bool bi
 		case ConvertedType::MAP:
 		case ConvertedType::MAP_KEY_VALUE:
 		case ConvertedType::LIST:
-		case ConvertedType::ENUM:
 		case ConvertedType::JSON:
 		case ConvertedType::BSON:
 		default:

data/parquet-testing/enum.parquet

@@ -4,8 +4,9 @@
 
 namespace duckdb {
 
-ProgressBar::ProgressBar(Executor &executor, idx_t show_progress_after)
-    : executor(executor), show_progress_after(show_progress_after), current_percentage(-1) {
+ProgressBar::ProgressBar(Executor &executor, idx_t show_progress_after, bool print_progress)
+    : executor(executor), show_progress_after(show_progress_after), current_percentage(-1),
+      print_progress(print_progress) {
 }
 
 double ProgressBar::GetCurrentPercentage() {
@@ -28,7 +29,6 @@ void ProgressBar::Update(bool final) {
 		return;
 	}
 	auto sufficient_time_elapsed = profiler.Elapsed() > show_progress_after / 1000.0;
-	auto print_progress = ClientConfig::GetConfig(executor.context).print_progress_bar;
 	if (new_percentage > current_percentage) {
 		current_percentage = new_percentage;
 	}

extension/parquet/parquet_reader.cpp

@@ -304,28 +304,23 @@ idx_t GroupedAggregateHashTable::AddChunk(DataChunk &groups, Vector &group_hashe
 			// value have not been seen yet
 			idx_t new_group_count =
 			    distinct_hashes[aggr_idx]->FindOrCreateGroups(probe_chunk, dummy_addresses, new_groups);
-
-			// now fix up the payload and addresses accordingly by creating
-			// a selection vector
 			if (new_group_count > 0) {
+				// now fix up the payload and addresses accordingly by creating
+				// a selection vector
+				DataChunk distinct_payload;
+				distinct_payload.Initialize(payload.GetTypes());
+				distinct_payload.Slice(payload, new_groups, new_group_count);
+				distinct_payload.Verify();
+
+				Vector distinct_addresses(addresses, new_groups, new_group_count);
+				distinct_addresses.Verify(new_group_count);
+
 				if (aggr.filter) {
-					Vector distinct_addresses(addresses, new_groups, new_group_count);
-					DataChunk distinct_payload;
-					auto pay_types = payload.GetTypes();
-					distinct_payload.Initialize(pay_types);
-					distinct_payload.Slice(payload, new_groups, new_group_count);
-					distinct_addresses.Verify(new_group_count);
 					distinct_addresses.Normalify(new_group_count);
 					RowOperations::UpdateFilteredStates(aggr, distinct_addresses, distinct_payload, payload_idx);
 				} else {
-					Vector distinct_addresses(addresses, new_groups, new_group_count);
-					for (idx_t i = 0; i < aggr.child_count; i++) {
-						payload.data[payload_idx + i].Slice(new_groups, new_group_count);
-						payload.data[payload_idx + i].Verify(new_group_count);
-					}
-					distinct_addresses.Verify(new_group_count);
-
-					RowOperations::UpdateStates(aggr, distinct_addresses, payload, payload_idx, new_group_count);
+					RowOperations::UpdateStates(aggr, distinct_addresses, distinct_payload, payload_idx,
+					                            new_group_count);
 				}
 			}
 		} else if (aggr.filter) {

src/common/progress_bar.cpp

@@ -161,6 +161,7 @@ SinkResultType PhysicalHashAggregate::Sink(ExecutionContext &context, GlobalSink
 		for (auto &child_expr : aggr.children) {
 			D_ASSERT(child_expr->type == ExpressionType::BOUND_REF);
 			auto &bound_ref_expr = (BoundReferenceExpression &)*child_expr;
+			D_ASSERT(bound_ref_expr.index < input.data.size());
 			aggregate_input_chunk.data[aggregate_input_idx++].Reference(input.data[bound_ref_expr.index]);
 		}
 	}
@@ -169,6 +170,7 @@ SinkResultType PhysicalHashAggregate::Sink(ExecutionContext &context, GlobalSink
 		if (aggr.filter) {
 			auto it = filter_indexes.find(aggr.filter.get());
 			D_ASSERT(it != filter_indexes.end());
+			D_ASSERT(it->second < input.data.size());
 			aggregate_input_chunk.data[aggregate_input_idx++].Reference(input.data[it->second]);
 		}
 	}

src/execution/aggregate_hashtable.cpp

@@ -901,6 +901,25 @@ static void ComputeWindowExpression(BoundWindowExpression *wexpr, ChunkCollectio
 		}
 	}
 
+	// evaluate the FILTER clause and stuff it into a large mask for compactness and reuse
+	ValidityMask filter_mask;
+	vector<validity_t> filter_bits;
+	if (wexpr->filter_expr) {
+		// 	Start with all invalid and set the ones that pass
+		filter_bits.resize(ValidityMask::ValidityMaskSize(input.Count()), 0);
+		filter_mask.Initialize(filter_bits.data());
+		ExpressionExecutor filter_execution(*wexpr->filter_expr);
+		SelectionVector true_sel(STANDARD_VECTOR_SIZE);
+		idx_t base_idx = 0;
+		for (auto &chunk : input.Chunks()) {
+			const auto filtered = filter_execution.SelectExpression(*chunk, true_sel);
+			for (idx_t f = 0; f < filtered; ++f) {
+				filter_mask.SetValid(base_idx + true_sel[f]);
+			}
+			base_idx += chunk->size();
+		}
+	}
+
 	// evaluate boundaries if present. Parser has checked boundary types.
 	ChunkCollection boundary_start_collection;
 	if (wexpr->start_expr) {
@@ -954,7 +973,7 @@ static void ComputeWindowExpression(BoundWindowExpression *wexpr, ChunkCollectio
 
 	if (wexpr->aggregate) {
 		segment_tree = make_unique<WindowSegmentTree>(*(wexpr->aggregate), wexpr->bind_info.get(), wexpr->return_type,
-		                                              &payload_collection, mode);
+		                                              &payload_collection, filter_mask, mode);
 	}
 
 	WindowBoundariesState bounds(wexpr);

src/execution/operator/aggregate/physical_hash_aggregate.cpp

@@ -113,6 +113,14 @@ void PhysicalTableScan::GetData(ExecutionContext &context, DataChunk &chunk, Glo
 	}
 }
 
+double PhysicalTableScan::GetProgress(ClientContext &context, GlobalSourceState &gstate_p) const {
+	if (function.table_scan_progress) {
+		return function.table_scan_progress(context, bind_data.get());
+	}
+	// if table_scan_progress is not implemented we don't support this function yet in the progress bar
+	return -1;
+}
+
 idx_t PhysicalTableScan::GetBatchIndex(ExecutionContext &context, DataChunk &chunk, GlobalSourceState &gstate_p,
                                        LocalSourceState &lstate) const {
 	D_ASSERT(SupportsBatchIndex());

src/execution/operator/aggregate/physical_window.cpp

@@ -74,6 +74,10 @@ idx_t PhysicalOperator::GetBatchIndex(ExecutionContext &context, DataChunk &chun
                                       LocalSourceState &lstate) const {
 	throw InternalException("Calling GetBatchIndex on a node that does not support it");
 }
+
+double PhysicalOperator::GetProgress(ClientContext &context, GlobalSourceState &gstate) const {
+	return -1;
+}
 // LCOV_EXCL_STOP
 
 //===--------------------------------------------------------------------===//

src/execution/operator/scan/physical_table_scan.cpp

@@ -8,17 +8,19 @@ namespace duckdb {
 
 WindowSegmentTree::WindowSegmentTree(AggregateFunction &aggregate, FunctionData *bind_info,
                                      const LogicalType &result_type_p, ChunkCollection *input,
-                                     WindowAggregationMode mode_p)
+                                     const ValidityMask &filter_mask_p, WindowAggregationMode mode_p)
     : aggregate(aggregate), bind_info(bind_info), result_type(result_type_p), state(aggregate.state_size()),
       statep(Value::POINTER((idx_t)state.data())), frame(0, 0), active(0, 1),
-      statev(Value::POINTER((idx_t)state.data())), internal_nodes(0), input_ref(input), mode(mode_p) {
+      statev(Value::POINTER((idx_t)state.data())), internal_nodes(0), input_ref(input), filter_mask(filter_mask_p),
+      mode(mode_p) {
 #if STANDARD_VECTOR_SIZE < 512
 	throw NotImplementedException("Window functions are not supported for vector sizes < 512");
 #endif
 	statep.Normalify(STANDARD_VECTOR_SIZE);
 	statev.SetVectorType(VectorType::FLAT_VECTOR); // Prevent conversion of results to constants
 
 	if (input_ref && input_ref->ColumnCount() > 0) {
+		filter_sel.Initialize(STANDARD_VECTOR_SIZE);
 		inputs.Initialize(input_ref->Types());
 		// if we have a frame-by-frame method, share the single state
 		if (aggregate.window && UseWindowAPI()) {
@@ -99,6 +101,19 @@ void WindowSegmentTree::ExtractFrame(idx_t begin, idx_t end) {
 			VectorOperations::Copy(chunk_b.data[i], v, chunk_b_count, 0, chunk_a_count);
 		}
 	}
+
+	// Slice to any filtered rows
+	if (!filter_mask.AllValid()) {
+		idx_t filtered = 0;
+		for (idx_t i = begin; i < end; ++i) {
+			if (filter_mask.RowIsValid(i)) {
+				filter_sel.set_index(filtered++, i - begin);
+			}
+		}
+		if (filtered != inputs.size()) {
+			inputs.Slice(filter_sel, filtered);
+		}
+	}
 }
 
 void WindowSegmentTree::WindowSegmentValue(idx_t l_idx, idx_t begin, idx_t end) {
@@ -179,7 +194,16 @@ void WindowSegmentTree::Compute(Vector &result, idx_t rid, idx_t begin, idx_t en
 	if (inputs.ColumnCount() == 0) {
 		D_ASSERT(GetTypeIdSize(result_type.InternalType()) == sizeof(idx_t));
 		auto data = FlatVector::GetData<idx_t>(result);
-		data[rid] = end - begin;
+		// Slice to any filtered rows
+		if (!filter_mask.AllValid()) {
+			idx_t filtered = 0;
+			for (idx_t i = begin; i < end; ++i) {
+				filtered += filter_mask.RowIsValid(i);
+			}
+			data[rid] = filtered;
+		} else {
+			data[rid] = end - begin;
+		}
 		return;
 	}
 
@@ -220,8 +244,8 @@ void WindowSegmentTree::Compute(Vector &result, idx_t rid, idx_t begin, idx_t en
 		active = FrameBounds(active_chunks.first * STANDARD_VECTOR_SIZE,
 		                     MinValue((active_chunks.second + 1) * STANDARD_VECTOR_SIZE, coll.Count()));
 
-		aggregate.window(inputs.data.data(), bind_info, inputs.ColumnCount(), state.data(), frame, prev, result, rid,
-		                 active.first);
+		aggregate.window(inputs.data.data(), filter_mask, bind_info, inputs.ColumnCount(), state.data(), frame, prev,
+		                 result, rid, active.first);
 		return;
 	}