{
  "plan" : [ {
    "@type" : "ksqlPlanV1",
    "statementText" : "CREATE STREAM INPUT_STREAM (K STRING KEY, SF BIGINT) WITH (KAFKA_TOPIC='stream_topic', VALUE_FORMAT='JSON');",
    "ddlCommand" : {
      "@type" : "createStreamV1",
      "sourceName" : "INPUT_STREAM",
      "schema" : "`K` STRING KEY, `SF` BIGINT",
      "topicName" : "stream_topic",
      "formats" : {
        "keyFormat" : {
          "format" : "KAFKA"
        },
        "valueFormat" : {
          "format" : "JSON"
        }
      }
    }
  }, {
    "@type" : "ksqlPlanV1",
    "statementText" : "CREATE TABLE INPUT_TABLE (ID BIGINT PRIMARY KEY, TF INTEGER) WITH (KAFKA_TOPIC='table_topic', VALUE_FORMAT='JSON');",
    "ddlCommand" : {
      "@type" : "createTableV1",
      "sourceName" : "INPUT_TABLE",
      "schema" : "`ID` BIGINT KEY, `TF` INTEGER",
      "topicName" : "table_topic",
      "formats" : {
        "keyFormat" : {
          "format" : "KAFKA"
        },
        "valueFormat" : {
          "format" : "JSON"
        }
      }
    }
  }, {
    "@type" : "ksqlPlanV1",
    "statementText" : "CREATE STREAM OUTPUT AS SELECT\n  *,\n  S.ROWTIME S_ROWTIME,\n  T.ROWTIME T_ROWTIME\nFROM INPUT_STREAM S\nINNER JOIN INPUT_TABLE T ON ((S.SF = T.ID))\nEMIT CHANGES",
    "ddlCommand" : {
      "@type" : "createStreamV1",
      "sourceName" : "OUTPUT",
      "schema" : "`S_SF` BIGINT KEY, `S_K` STRING, `T_ID` BIGINT, `T_TF` INTEGER, `S_ROWTIME` BIGINT, `T_ROWTIME` BIGINT",
      "topicName" : "OUTPUT",
      "formats" : {
        "keyFormat" : {
          "format" : "KAFKA"
        },
        "valueFormat" : {
          "format" : "JSON"
        }
      }
    },
    "queryPlan" : {
      "sources" : [ "INPUT_STREAM", "INPUT_TABLE" ],
      "sink" : "OUTPUT",
      "physicalPlan" : {
        "@type" : "streamSinkV1",
        "properties" : {
          "queryContext" : "OUTPUT"
        },
        "source" : {
          "@type" : "streamSelectV1",
          "properties" : {
            "queryContext" : "Project"
          },
          "source" : {
            "@type" : "streamTableJoinV1",
            "properties" : {
              "queryContext" : "Join"
            },
            "joinType" : "INNER",
            "internalFormats" : {
              "keyFormat" : {
                "format" : "KAFKA"
              },
              "valueFormat" : {
                "format" : "JSON"
              }
            },
            "leftSource" : {
              "@type" : "streamSelectV1",
              "properties" : {
                "queryContext" : "PrependAliasLeft"
              },
              "source" : {
                "@type" : "streamSelectKeyV2",
                "properties" : {
                  "queryContext" : "LeftSourceKeyed"
                },
                "source" : {
                  "@type" : "streamSourceV1",
                  "properties" : {
                    "queryContext" : "KafkaTopic_Left/Source"
                  },
                  "topicName" : "stream_topic",
                  "formats" : {
                    "keyFormat" : {
                      "format" : "KAFKA"
                    },
                    "valueFormat" : {
                      "format" : "JSON"
                    }
                  },
                  "sourceSchema" : "`K` STRING KEY, `SF` BIGINT"
                },
                "keyExpression" : "SF"
              },
              "keyColumnNames" : [ "S_SF" ],
              "selectExpressions" : [ "SF AS S_SF", "ROWTIME AS S_ROWTIME", "K AS S_K" ]
            },
            "rightSource" : {
              "@type" : "tableSelectV1",
              "properties" : {
                "queryContext" : "PrependAliasRight"
              },
              "source" : {
                "@type" : "tableSourceV1",
                "properties" : {
                  "queryContext" : "KafkaTopic_Right/Source"
                },
                "topicName" : "table_topic",
                "formats" : {
                  "keyFormat" : {
                    "format" : "KAFKA"
                  },
                  "valueFormat" : {
                    "format" : "JSON"
                  }
                },
                "sourceSchema" : "`ID` BIGINT KEY, `TF` INTEGER"
              },
              "keyColumnNames" : [ "T_ID" ],
              "selectExpressions" : [ "TF AS T_TF", "ROWTIME AS T_ROWTIME", "ID AS T_ID" ]
            },
            "keyColName" : "S_SF"
          },
          "keyColumnNames" : [ "S_SF" ],
          "selectExpressions" : [ "S_K AS S_K", "T_ID AS T_ID", "T_TF AS T_TF", "S_ROWTIME AS S_ROWTIME", "T_ROWTIME AS T_ROWTIME" ]
        },
        "formats" : {
          "keyFormat" : {
            "format" : "KAFKA"
          },
          "valueFormat" : {
            "format" : "JSON"
          }
        },
        "topicName" : "OUTPUT"
      },
      "queryId" : "CSAS_OUTPUT_0"
    }
  } ],
  "configs" : {
    "ksql.extension.dir" : "ext",
    "ksql.streams.cache.max.bytes.buffering" : "0",
    "ksql.security.extension.class" : null,
    "ksql.transient.prefix" : "transient_",
    "ksql.persistence.wrap.single.values" : "true",
    "ksql.authorization.cache.expiry.time.secs" : "30",
    "ksql.schema.registry.url" : "",
    "ksql.streams.default.deserialization.exception.handler" : "io.confluent.ksql.errors.LogMetricAndContinueExceptionHandler",
    "ksql.output.topic.name.prefix" : "",
    "ksql.streams.auto.offset.reset" : "earliest",
    "ksql.query.pull.enable.standby.reads" : "false",
    "ksql.connect.url" : "http://localhost:8083",
    "ksql.service.id" : "some.ksql.service.id",
    "ksql.internal.topic.min.insync.replicas" : "1",
    "ksql.streams.shutdown.timeout.ms" : "300000",
    "ksql.internal.topic.replicas" : "1",
    "ksql.insert.into.values.enabled" : "true",
    "ksql.query.pull.max.allowed.offset.lag" : "9223372036854775807",
    "ksql.query.pull.max.qps" : "2147483647",
    "ksql.streams.default.production.exception.handler" : "io.confluent.ksql.errors.ProductionExceptionHandlerUtil$LogAndFailProductionExceptionHandler",
    "ksql.access.validator.enable" : "auto",
    "ksql.streams.bootstrap.servers" : "localhost:0",
    "ksql.streams.commit.interval.ms" : "2000",
    "ksql.metric.reporters" : "",
    "ksql.query.pull.metrics.enabled" : "false",
    "ksql.streams.auto.commit.interval.ms" : "0",
    "ksql.metrics.extension" : null,
    "ksql.streams.topology.optimization" : "all",
    "ksql.hidden.topics" : "_confluent.*,__confluent.*,_schemas,__consumer_offsets,__transaction_state,connect-configs,connect-offsets,connect-status,connect-statuses",
    "ksql.streams.num.stream.threads" : "4",
    "ksql.timestamp.throw.on.invalid" : "false",
    "ksql.authorization.cache.max.entries" : "10000",
    "ksql.metrics.tags.custom" : "",
    "ksql.pull.queries.enable" : "true",
    "ksql.udfs.enabled" : "true",
    "ksql.udf.enable.security.manager" : "true",
    "ksql.connect.worker.config" : "",
    "ksql.sink.window.change.log.additional.retention" : "1000000",
    "ksql.readonly.topics" : "_confluent.*,__confluent.*,_schemas,__consumer_offsets,__transaction_state,connect-configs,connect-offsets,connect-status,connect-statuses",
    "ksql.udf.collect.metrics" : "false",
    "ksql.persistent.prefix" : "query_",
    "ksql.query.persistent.active.limit" : "2147483647"
  }
}