Quack node

Makefile

@@ -9,9 +9,10 @@ SRCS = src/quack_detoast.cpp \
 	   src/quack_heap_scan.cpp \
 	   src/quack_heap_seq_scan.cpp \
 	   src/quack_hooks.cpp \
-	   src/quack_select.cpp \
-	   src/quack_types.cpp \
 	   src/quack_memory_allocator.cpp \
+	   src/quack_node.cpp \
+	   src/quack_planner.cpp \
+	   src/quack_types.cpp \
 	   src/quack.cpp
 
 OBJS = $(subst .cpp,.o, $(SRCS))
@@ -54,7 +55,7 @@ ifeq ($(UNAME_S),Linux)
 	DUCKDB_LIB = libduckdb.so
 endif
 
-all: duckdb $(OBJS)
+all: duckdb $(OBJS) .depend
 
 include $(PGXS)
 
@@ -64,7 +65,7 @@ third_party/duckdb/Makefile:
 	git submodule update --init --recursive
 
 third_party/duckdb/build/$(QUACK_BUILD_DUCKDB)/src/$(DUCKDB_LIB):
-	$(MAKE) -C third_party/duckdb $(QUACK_BUILD_DUCKDB) DISABLE_SANITIZER=1 ENABLE_UBSAN=0 BUILD_UNITTESTS=OFF CMAKE_EXPORT_COMPILE_COMMANDS=1
+	$(MAKE) -C third_party/duckdb $(QUACK_BUILD_DUCKDB) DISABLE_SANITIZER=1 ENABLE_UBSAN=0 BUILD_UNITTESTS=OFF BUILD_HTTPFS=1 CMAKE_EXPORT_COMPILE_COMMANDS=1
 
 install_duckdb:
 	$(install_bin) -m 755 third_party/duckdb/build/$(QUACK_BUILD_DUCKDB)/src/$(DUCKDB_LIB) $(DESTDIR)$(PG_LIB)

expected/basic.out

@@ -4,41 +4,45 @@ INSERT INTO t SELECT g % 10 from generate_series(1,1000000) g;
 SET client_min_messages to 'DEBUG3';
 SELECT COUNT(*) FROM t;
 DEBUG:  -- (DuckDB/PostgresHeapBind) Column name: a, Type: INTEGER --
+DEBUG:  -- (DuckDB/PostgresHeapBind) Column name: a, Type: INTEGER --
 DEBUG:  -- (DuckDB/PostgresHeapScanGlobalState) Running 1 threads -- 
-  count  
----------
- 1000000
+ count_star() 
+--------------
+      1000000
 (1 row)
 
 SELECT a, COUNT(*) FROM t WHERE a > 5 GROUP BY a ORDER BY a;
 DEBUG:  -- (DuckDB/PostgresHeapBind) Column name: a, Type: INTEGER --
+DEBUG:  -- (DuckDB/PostgresHeapBind) Column name: a, Type: INTEGER --
 DEBUG:  -- (DuckDB/PostgresHeapScanGlobalState) Running 1 threads -- 
- a | count  
----+--------
- 6 | 100000
- 7 | 100000
- 8 | 100000
- 9 | 100000
+ a | count_star() 
+---+--------------
+ 6 |       100000
+ 7 |       100000
+ 8 |       100000
+ 9 |       100000
 (4 rows)
 
 SET quack.max_threads_per_query to 4;
 SELECT COUNT(*) FROM t;
 DEBUG:  -- (DuckDB/PostgresHeapBind) Column name: a, Type: INTEGER --
+DEBUG:  -- (DuckDB/PostgresHeapBind) Column name: a, Type: INTEGER --
 DEBUG:  -- (DuckDB/PostgresHeapScanGlobalState) Running 4 threads -- 
-  count  
----------
- 1000000
+ count_star() 
+--------------
+      1000000
 (1 row)
 
 SELECT a, COUNT(*) FROM t WHERE a > 5 GROUP BY a ORDER BY a;
 DEBUG:  -- (DuckDB/PostgresHeapBind) Column name: a, Type: INTEGER --
+DEBUG:  -- (DuckDB/PostgresHeapBind) Column name: a, Type: INTEGER --
 DEBUG:  -- (DuckDB/PostgresHeapScanGlobalState) Running 4 threads -- 
- a | count  
----+--------
- 6 | 100000
- 7 | 100000
- 8 | 100000
- 9 | 100000
+ a | count_star() 
+---+--------------
+ 6 |       100000
+ 7 |       100000
+ 8 |       100000
+ 9 |       100000
 (4 rows)
 
 SET quack.max_threads_per_query TO default;

include/quack/quack_filter.hpp

@@ -7,5 +7,7 @@ extern "C" {
 }
 
 namespace quack {
+
 bool ApplyValueFilter(duckdb::TableFilter &filter, Datum &value, bool isNull, Oid typeOid);
+
 } // namespace quack

include/quack/quack_heap_scan.hpp

@@ -78,12 +78,13 @@ struct PostgresHeapScanFunction : public duckdb::TableFunction {
 
 struct PostgresHeapReplacementScanData : public duckdb::ReplacementScanData {
 public:
-	PostgresHeapReplacementScanData(QueryDesc *desc) : desc(desc) {
+	PostgresHeapReplacementScanData(Query *parse, const char *query) : m_parse(parse), m_query(query) {
 	}
 	~PostgresHeapReplacementScanData() override {};
 
 public:
-	QueryDesc *desc;
+	Query *m_parse;
+	std::string m_query;
 };
 
 duckdb::unique_ptr<duckdb::TableRef> PostgresHeapReplacementScan(duckdb::ClientContext &context,

include/quack/quack_heap_seq_scan.hpp

@@ -79,6 +79,7 @@ class PostgresHeapSeqScan {
 
 public:
 	Relation GetRelation();
+	void CloseRelation();
 	TupleDesc GetTupleDesc();
 	bool ReadPageTuples(duckdb::DataChunk &output, PostgresHeapSeqScanThreadInfo &threadScanInfo);
 	bool IsValid() const;
@@ -87,6 +88,7 @@ class PostgresHeapSeqScan {
 	Page PreparePageRead(PostgresHeapSeqScanThreadInfo &threadScanInfo);
 
 private:
+	RangeTblEntry * m_tableEntry = nullptr;
 	Relation m_rel = nullptr;
 	Snapshot m_snapshot = nullptr;
 	PostgresHeapSeqParallelScanState m_parallel_scan_state;

include/quack/quack_node.hpp

@@ -0,0 +1,9 @@
+#pragma once
+
+extern "C" {
+#include "postgres.h"
+#include "nodes/extensible.h"
+}
+
+extern CustomScanMethods quack_scan_scan_methods;
+extern "C" void quack_init_node(void);

include/quack/quack_planner.hpp

@@ -0,0 +1,8 @@
+#pragma once
+
+extern "C" {
+#include "postgres.h"
+#include "optimizer/planner.h"
+}
+
+PlannedStmt *quack_plan_node(Query *parse, const char *query_string, int cursorOptions, ParamListInfo boundParams);

include/quack/quack_types.hpp

@@ -10,9 +10,16 @@ extern "C" {
 #include "quack/quack_heap_seq_scan.hpp"
 
 namespace quack {
+
+// DuckDB has date starting from 1/1/1970 while PG starts from 1/1/2000
+constexpr int32_t QUACK_DUCK_DATE_OFFSET = 10957;
+constexpr int64_t QUACK_DUCK_TIMESTAMP_OFFSET = INT64CONST(10957) * USECS_PER_DAY;
+
 duckdb::LogicalType ConvertPostgresToDuckColumnType(Oid type);
+Oid GetPostgresDuckDBType(duckdb::LogicalTypeId type);
 void ConvertPostgresToDuckValue(Datum value, duckdb::Vector &result, idx_t offset);
 void ConvertDuckToPostgresValue(TupleTableSlot *slot, duckdb::Value &value, idx_t col);
 void InsertTupleIntoChunk(duckdb::DataChunk &output, PostgresHeapSeqScanThreadInfo &threadScanInfo,
                           PostgresHeapSeqParallelScanState &parallelScanState);
+
 } // namespace quack

quack--0.0.1.sql

@@ -1 +1,9 @@
 LOAD 'quack';
+
+CREATE OR REPLACE FUNCTION read_parquet(path text)
+RETURNS SETOF record LANGUAGE 'plpgsql' AS
+$func$
+BEGIN
+  RETURN QUERY EXECUTE 'SELECT 1';
+END;
+$func$;

src/quack.cpp

@@ -1,10 +1,10 @@
 extern "C" {
 #include "postgres.h"
-
 #include "utils/guc.h"
 }
 
 #include "quack/quack.h"
+#include "quack/quack_node.hpp"
 
 static void quack_init_guc(void);
 
@@ -17,6 +17,7 @@ void
 _PG_init(void) {
 	quack_init_guc();
 	quack_init_hooks();
+	quack_init_node();
 }
 }
 

src/quack_filter.cpp

@@ -6,6 +6,9 @@ extern "C" {
 #include "catalog/pg_type.h"
 }
 
+#include "quack/quack_filter.hpp"
+#include "quack/quack_types.hpp"
+
 namespace quack {
 
 template <class T, class OP>
@@ -33,8 +36,24 @@ FilterOperationSwitch(Datum &value, duckdb::Value &constant, Oid typeOid) {
 	case INT8OID:
 		return TemplatedFilterOperation<int64_t, OP>(value, constant);
 		break;
+	case FLOAT4OID:
+		return TemplatedFilterOperation<float, OP>(value, constant);
+		break;
+	case FLOAT8OID:
+		return TemplatedFilterOperation<double, OP>(value, constant);
+		break;
+	case DATEOID: {
+		Datum dateDatum = static_cast<int32_t>(value + quack::QUACK_DUCK_DATE_OFFSET);
+		return TemplatedFilterOperation<int32_t, OP>(dateDatum, constant);
+		break;
+	}
+	case TIMESTAMPOID: {
+		Datum timeStampDatum = static_cast<int64_t>(value + quack::QUACK_DUCK_TIMESTAMP_OFFSET);
+		return TemplatedFilterOperation<int64_t, OP>(timeStampDatum, constant);
+		break;
+	}
 	default:
-		elog(ERROR, "Unsupported quack type: %d", typeOid);
+		elog(ERROR, "(DuckDB/FilterOperationSwitch) Unsupported quack type: %d", typeOid);
 	}
 }
 

src/quack_heap_scan.cpp

@@ -188,14 +188,14 @@ PostgresHeapReplacementScan(duckdb::ClientContext &context, const duckdb::string
 	auto &scan_data = reinterpret_cast<PostgresHeapReplacementScanData &>(*data);
 
 	/* Check name against query rtable list and verify that it is heap table */
-	auto table = FindMatchingHeapRelation(scan_data.desc->plannedstmt->rtable, table_name);
+	auto table = FindMatchingHeapRelation(scan_data.m_parse->rtable, table_name);
 
 	if (!table) {
 		return nullptr;
 	}
 
 	// Create POINTER values from the 'table' and 'snapshot' variables
-	auto children = CreateFunctionArguments(table, scan_data.desc->estate->es_snapshot);
+	auto children = CreateFunctionArguments(table, GetActiveSnapshot());
 	auto table_function = duckdb::make_uniq<duckdb::TableFunctionRef>();
 	table_function->function = duckdb::make_uniq<duckdb::FunctionExpression>("postgres_heap_scan", std::move(children));
 

src/quack_heap_seq_scan.cpp

@@ -14,9 +14,8 @@ extern "C" {
 #include <thread>
 
 namespace quack {
-
 PostgresHeapSeqScan::PostgresHeapSeqScan(RangeTblEntry *table)
-    : m_rel(RelationIdGetRelation(table->relid)), m_snapshot(nullptr) {
+    : m_tableEntry(table), m_rel(nullptr), m_snapshot(nullptr) {
 }
 
 PostgresHeapSeqScan::~PostgresHeapSeqScan() {
@@ -25,15 +24,28 @@ PostgresHeapSeqScan::~PostgresHeapSeqScan() {
 	}
 }
 
-PostgresHeapSeqScan::PostgresHeapSeqScan(PostgresHeapSeqScan &&other) : m_rel(other.m_rel) {
-	other.m_rel = nullptr;
+PostgresHeapSeqScan::PostgresHeapSeqScan(PostgresHeapSeqScan &&other)
+    : m_tableEntry(other.m_tableEntry), m_rel(nullptr) {
+	other.CloseRelation();
+	other.m_tableEntry = nullptr;
 }
 
 Relation
 PostgresHeapSeqScan::GetRelation() {
+	if (m_tableEntry && m_rel == nullptr) {
+		m_rel = RelationIdGetRelation(m_tableEntry->relid);
+	}
 	return m_rel;
 }
 
+void
+PostgresHeapSeqScan::CloseRelation() {
+	if (IsValid()) {
+		RelationClose(m_rel);
+	}
+	m_rel = nullptr;
+}
+
 bool
 PostgresHeapSeqScan::IsValid() const {
 	return RelationIsValid(m_rel);
@@ -56,7 +68,8 @@ PostgresHeapSeqScan::PreparePageRead(PostgresHeapSeqScanThreadInfo &threadScanIn
 }
 
 void
-PostgresHeapSeqScan::InitParallelScanState( duckdb::TableFunctionInitInput &input) {
+PostgresHeapSeqScan::InitParallelScanState(duckdb::TableFunctionInitInput &input) {
+	(void) GetRelation();
 	m_parallel_scan_state.m_nblocks = RelationGetNumberOfBlocks(m_rel);
 
 	/* SELECT COUNT(*) FROM */
@@ -72,16 +85,15 @@ PostgresHeapSeqScan::InitParallelScanState( duckdb::TableFunctionInitInput &inpu
 
 	if (input.CanRemoveFilterColumns()) {
 		for (duckdb::idx_t i = 0; i < input.projection_ids.size(); i++) {
-			m_parallel_scan_state.m_projections[input.projection_ids[i]] = input.column_ids[i];
+			m_parallel_scan_state.m_projections[i] = input.column_ids[input.projection_ids[i]];
 		}
 	} else {
 		for (duckdb::idx_t i = 0; i < input.projection_ids.size(); i++) {
 			m_parallel_scan_state.m_projections[i] = input.column_ids[i];
 		}
 	}
 
-
-	//m_parallel_scan_state.PrefetchNextRelationPages(m_rel);
+	// m_parallel_scan_state.PrefetchNextRelationPages(m_rel);
 	m_parallel_scan_state.m_filters = input.filters.get();
 }
 
@@ -110,7 +122,7 @@ PostgresHeapSeqScan::ReadPageTuples(duckdb::DataChunk &output, PostgresHeapSeqSc
 			threadScanInfo.m_buffer =
 			    ReadBufferExtended(m_rel, MAIN_FORKNUM, block, RBM_NORMAL, m_parallel_scan_state.m_strategy);
 			LockBuffer(threadScanInfo.m_buffer, BUFFER_LOCK_SHARE);
-			//m_parallel_scan_state.PrefetchNextRelationPages(m_rel);
+			// m_parallel_scan_state.PrefetchNextRelationPages(m_rel);
 			m_parallel_scan_state.m_lock.unlock();
 			page = PreparePageRead(threadScanInfo);
 			threadScanInfo.m_read_next_page = false;
@@ -196,7 +208,6 @@ PostgresHeapSeqParallelScanState::PrefetchNextRelationPages(Relation rel) {
 	    (m_last_prefetch_block - m_last_assigned_block_number) > 8)
 		return;
 
-
 	for (BlockNumber i = m_last_prefetch_block; i < last_batch_prefetch_block_num; i++) {
 		PrefetchBuffer(rel, MAIN_FORKNUM, m_last_prefetch_block);
 		m_last_prefetch_block++;