Jooq 雪花复制命令供应商特定功能

Question

有没有人使用 sql 模板或其他方法在 JOOQ 中实现了雪花的 COPY 命令？我在那里找不到任何东西。我在 JOOQ 的 github 页面上看到了一个关于它的问题，但目前似乎没有支持 COPY 命令的计划，因为它是供应商特定的功能。 如果那里什么都没有，也许有一些关于支持它的最佳方式的提示。围绕它的语法有很多不同的选项。

Answer 1

我认为这里的一些中间立场将是您实现动态COPY 语句支持的最佳选择。当然，你可以实现一个成熟的 jOOQ 风格的 mini-DSL 来支持COPY。 Here's the main idea behind jOOQ's fluent API design。但我认为这对于最常见的用例来说可能有点过头了。

Leveraging plain SQL templating 带有一些重载可能就足够了。 Looking at the Snowflake manual section about COPY，这可能是一个最小的解决方案：

public static Query copy(
    DSLContext ctx,
    Table<?> into,
    QueryPart from,
    List<String> files
) {
    return ctx.query("copy into {0} from {1} files = ({2})", 
        table, 
        from,
        DSL.list(files.stream().map(DSL::inline).toList())
    );

当然，您可以根据需要将其设为多功能和动态，提供更多重载，或者再次提供一些构建器 API 甚至 DSL。还可以查看CustomQueryPart 和其他custom syntax elements，因为当字符串模板还不够时。

Answer 2

我更进一步并实现了一个 DSL，具体语法为：COPY INTO <table/stage> FROM (SELECT FROM <table/stage>)...，这是我们的用例。
下面是 DSL 的主要代码，稍加简化。您将看到对未显示 impl 的各种自定义元素的引用，主要是它们扩展 CustomQueryPart。

请注意，它在 Scala 中 - 如果您不熟悉，请提前道歉。当我找到时间我将编辑并转换为 Java：

trait SnowflakeDSLContext extends DSLContext {
    def copyInto[R <: Record, T1, T2](table: Table[R],
                                      f1: Field[T1],
                                      f2: Field[T2]) : CopyIntoFromStep2[R, T1, T2]
}


class SnowflakeDSLContextImpl(config: Configuration)
    extends DefaultDSLContext(config) with SnowflakeDSLContext {

    def copyInto[R <: Record, T1, T2](table: Table[R], 
                                      f1: Field[T1],
                                      f2: Field[T2]): CopyIntoFromStep2[R, T1, T2] =
        new CopyIntoImpl[R, T1, T2](this, table, Seq(f1, f2))
 
   
trait CopyIntoFromStep2[R <: Record, T1, T2] {
    def from(select: Select[_ <: Record2[T1, T2]]): CopyIntoParametersStep
}


trait CopyIntoParametersStep extends CopyIntoFinalStep {
    def fileFormat(formatType: FileFormatType,
                   maybeComprType: Option[CompressionType] = None): CopyIntoParametersStep

    def purge(enabled: Boolean): CopyIntoParametersStep
}


trait CopyIntoFinalStep {
    def fetchResults(): Seq[CopyIntoResult]
}

// This is the main class which "does the trick" 
// by accumulating the query parts and building the final query
case class CopyIntoImpl[R <: Record, T1, T2](
    dslContext: DSLContext,
    table: Table[R],
    fields: Seq[Field[_]],
    maybeSelect: Option[Select[_ <: Record2[T1, T2]]] = None,
    maybeFileFormat: Option[FileFormat] = None,
    copyOptions: Seq[SingleValueParam[_]] = Nil
) extends CopyIntoFromStep2[R, T1, T2] with CopyIntoParametersStep {

    override def from(select: Select[_ <: Record2[T1, T2]]): CopyIntoParametersStep = {
        copy(maybeSelect = Some(select))
    }

    override def fileFormat(formatType: FileFormatType,
                            maybeCompr: Option[CompressionType] = None): CopyIntoParametersStep = {
        copy(maybeFileFormat = Some(FileFormat(formatType, maybeCompr)))
    }

    override def purge(enabled: Boolean): CopyIntoParametersStep =
        copy(copyOptions = copyOptions :+ Purge(enabled))

    override def fetchResults(): Seq[CopyIntoResult] = {
        dslContext.resultQuery("COPY INTO {0} {1} FROM ({2}) {3} {4}",
                               table,
                               UnqualifiedFieldList(fields),
                               maybeSelect.getOrElse(""),
                               maybeFileFormat.getOrElse(""),
                               LineDelimitedList(copyOptions))
                  .fetch(CopyIntoResultMapper)
                  .asScala
                  .toSeq
    }
}

用法示例：

val results = dslContext.copyInto(targetTable, field1, field2)
                        .from(DSL.select(field1, field2)
                                 .from(sourceTable))
                        .fileFormat(JSON, Some(GZIP))
                        .purge(true)
                        .fetchResults()