Use pg_lake list as default, fall back to crunchy if it does not exist

Author: sfc-gh-mslot

include/crunchy/incremental/file_list.h

@@ -1,11 +1,13 @@
 #pragma once
 
-#define DEFAULT_FILE_LIST_FUNCTION "crunchy_lake.list_files"
+#define DEFAULT_FILE_LIST_FUNCTION "lake_file.list"
+#define ORIGINAL_DEFAULT_FILE_LIST_FUNCTION "crunchy_lake.list_files"
 
 extern char *DefaultFileListFunction;
 
 void		InitializeFileListPipelineState(char *pipelineName, char *prefix, bool batched, char *listFunction, int maxBatchSize);
 void		RemoveProcessedFileList(char *pipelineName);
 void		ExecuteFileListPipeline(char *pipelineName, char *command);
+bool		ListFunctionExists(char *listFunction);
 char	   *SanitizeListFunction(char *listFunction);
 void		InsertProcessedFile(char *pipelineName, char *path);

src/file_list.c

@@ -541,6 +541,26 @@ RemoveProcessedFileList(char *pipelineName)
 }
 
 
+/*
+ * ListFunctionExists determines whether the given list function
+ * exists.
+ */
+bool
+ListFunctionExists(char *listFunction)
+{
+#if (PG_VERSION_NUM >= 160000)
+	List	   *names = stringToQualifiedNameList(listFunction, NULL);
+#else
+	List	   *names = stringToQualifiedNameList(listFunction);
+#endif
+	Oid			argTypes[] = {TEXTOID};
+	bool		missingOk = true;
+	Oid			functionId = LookupFuncName(names, 1, argTypes, missingOk);
+
+	return OidIsValid(functionId);
+}
+
+
 /*
  * SanitizeListFunction qualifies a list function name and errors
  * if the function cannot be found.

src/pipeline.c

@@ -221,13 +221,30 @@ incremental_create_file_list_pipeline(PG_FUNCTION_ARGS)
 	char	   *pipelineName = text_to_cstring(PG_GETARG_TEXT_P(0));
 	char	   *prefix = text_to_cstring(PG_GETARG_TEXT_P(1));
 	char	   *command = text_to_cstring(PG_GETARG_TEXT_P(2));
-	char	   *listFunction = PG_ARGISNULL(3) ? DefaultFileListFunction : text_to_cstring(PG_GETARG_TEXT_P(3));
+	char	   *listFunction = PG_ARGISNULL(3) ? NULL : text_to_cstring(PG_GETARG_TEXT_P(3));
 	bool		batched = PG_ARGISNULL(4) ? false : PG_GETARG_BOOL(4);
 	int			maxBatchSize = PG_ARGISNULL(5) ? 0 : PG_GETARG_INT32(5);
 	char	   *schedule = PG_ARGISNULL(6) ? NULL : text_to_cstring(PG_GETARG_TEXT_P(6));
 	bool		executeImmediately = PG_ARGISNULL(7) ? false : PG_GETARG_BOOL(7);
 	char	   *searchPath = pstrdup(namespace_search_path);
 
+	if (listFunction == NULL)
+	{
+		/*
+		 * Default to pg_lake, but fall back to original list function
+		 * if it does not exist for backwards compatibility.
+		 */
+		if (strcmp(DefaultFileListFunction, DEFAULT_FILE_LIST_FUNCTION) == 0 &&
+			!ListFunctionExists(DefaultFileListFunction))
+		{
+			listFunction = ORIGINAL_DEFAULT_FILE_LIST_FUNCTION;
+		}
+		else
+		{
+			listFunction = DefaultFileListFunction;
+		}
+	}
+
 	/* validate and sanitize function name */
 	listFunction = SanitizeListFunction(listFunction);