我有以下oracle sql查询：

SELECT * FROM 
    (SELECT SO_ORDER_KEY,QUEUE_TYPE,SYS_NO,
    DENSE_RANK() OVER (PARTITION BY SO_ORDER_KEY ORDER BY SYS_NO DESC) ORDER_RANK 
    FROM TSY940) 
WHERE ORDER_RANK=1;

在sqldeveloper中运行时，它返回所需的结果。
出于某种原因，当我在kafka connect jdbc属性中使用这个查询时

ERROR Failed to run query for table TimestampIncrementingTableQuerier{name='null', query='SELECT * FROM (SELECT SO_ORDER_KEY,QUEUE_TYPE,SYS_NO,DENSE_RANK() OVER (PARTITION BY SO_ORDER_KEY ORDER BY SYS_NO DESC) ORDER_RANK FROM TSY940) WHERE ORDER_RANK=1', topicPrefix='TSY940', timestampColumn='SYS_NO', incrementingColumn='null'}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask:247)
java.sql.SQLSyntaxErrorException: ORA-00933: SQL command not properly ended

        at oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:450)
        at oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:399)
        at oracle.jdbc.driver.T4C8Oall.processError(T4C8Oall.java:1059)
        at oracle.jdbc.driver.T4CTTIfun.receive(T4CTTIfun.java:522)
        at oracle.jdbc.driver.T4CTTIfun.doRPC(T4CTTIfun.java:257)
        at oracle.jdbc.driver.T4C8Oall.doOALL(T4C8Oall.java:587)
        at oracle.jdbc.driver.T4CPreparedStatement.doOall8(T4CPreparedStatement.java:225)
        at oracle.jdbc.driver.T4CPreparedStatement.doOall8(T4CPreparedStatement.java:53)
        at oracle.jdbc.driver.T4CPreparedStatement.executeForDescribe(T4CPreparedStatement.java:774)
        at oracle.jdbc.driver.OracleStatement.executeMaybeDescribe(OracleStatement.java:925)
        at oracle.jdbc.driver.OracleStatement.doExecuteWithTimeout(OracleStatement.java:1111)
        at oracle.jdbc.driver.OraclePreparedStatement.executeInternal(OraclePreparedStatement.java:4798)
        at oracle.jdbc.driver.OraclePreparedStatement.executeQuery(OraclePreparedStatement.java:4845)
        at oracle.jdbc.driver.OraclePreparedStatementWrapper.executeQuery(OraclePreparedStatementWrapper.java:1501)
        at io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.executeQuery(TimestampIncrementingTableQuerier.java:201)
        at io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:84)
        at io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.maybeStartQuery(TimestampIncrementingTableQuerier.java:55)
        at io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:225)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:179)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

这是我的属性文件：

name=poc-oracle-source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
connection.password =********
connection.url = jdbc:oracle:thin:@***.***.***.**:****/******
connection.user =***********
table.types=TABLE
query=SELECT * FROM (SELECT SO_ORDER_KEY,QUEUE_TYPE,SYS_NO,DENSE_RANK() OVER (PARTITION BY SO_ORDER_KEY ORDER BY SYS_NO DESC) ORDER_RANK FROM TSY940) WHERE ORDER_RANK=1
mode=timestamp
timestamp.column.name=SYS_NO
topic.prefix=TSY940
batch.max.rows = 500
poll.interval.ms=60000

transforms=createKey,extract
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.createKey.fields=SO_ORDER_KEY
transforms.extract.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extract.field=SO_ORDER_KEY

我使用ojdbc7驱动程序。
where子句可能是问题所在，因为当我将query属性替换为

query=SELECT * FROM (SELECT SO_ORDER_KEY,QUEUE_TYPE,SYS_NO,DENSE_RANK() OVER (PARTITION BY SO_ORDER_KEY ORDER BY SYS_NO DESC) ORDER_RANK FROM TSY940)

我知道这个问题有一个公认的答案，但是您遇到的问题（以及为什么将查询 Package 到另一个查询中解决了这个问题）是当连接作业正在运行时 timestamp 模式下，connect使用指定的时间戳字段向查询追加子句。这意味着connect将此无效查询传递给oracle：

SELECT *
    FROM (SELECT
            SO_ORDER_KEY,
            QUEUE_TYPE,
            SYS_NO,
            DENSE_RANK()
            OVER (
              PARTITION BY SO_ORDER_KEY
              ORDER BY SYS_NO DESC ) ORDER_RANK
          FROM TSY940)
    WHERE ORDER_RANK = 1
    WHERE SYS_NO >= {last_run_timestamp}

看到最后一行了吗？是由connect添加的。当您将查询 Package 到另一个 SELECT ，该附加子句不再引起问题。
当我第一次遇到这个问题时，我发现文档中没有提到这个问题，这让我非常恼火。

4条答案

按热度按时间

svujldwt1#

赞(0）回复(0）举报 2021-06-07

eyh26e7m2#

然后您可以尝试这个查询，它完全消除了 ORDER_RANK ```
SELECT SO_ORDER_KEY, QUEUE_TYPE, SYS_NO
FROM (
SELECT SO_ORDER_KEY, QUEUE_TYPE, SYS_NO
FROM (SELECT SO_ORDER_KEY, QUEUE_TYPE, SYS_NO,
DENSE_RANK() OVER(PARTITION BY SO_ORDER_KEY ORDER BY SYS_NO DESC) AS ORDER_RANK
FROM TSY940) sub
WHERE sub.ORDER_RANK=1
)

查看导致问题的原因的最佳方法是启用10046跟踪，并查看发送到数据库并导致错误的确切查询 `ORA-00933` .

kxeu7u2r3#

您可以尝试添加别名并删除 TO_CHAR :

SELECT SO_ORDER_KEY, QUEUE_TYPE, SYS_NO 
FROM (SELECT SO_ORDER_KEY, QUEUE_TYPE, SYS_NO,
      DENSE_RANK() OVER(PARTITION BY SO_ORDER_KEY ORDER BY SYS_NO DESC) AS ORDER_RANK 
      FROM TSY940) sub
WHERE sub.ORDER_RANK=1;

mbjcgjjk4#

失败，因为连接器附加了其timestamp子句。我们可以使用带有 predicate 的查询作为内部查询，例如：“select*from（select a，b，c from table where a='x'）。正如alex提到的，这是一个解决方法。请找出问题所在here:httpshttp://github.com/confluentinc/kafka-connect-jdbc/issues/566

00933:sql命令未正确结束

4条答案

相关问题

热门标签

最新问答