StreamCatalogService { @VisibleForTesting String translateFunctions(String sql, Map<String, String> udfInternalNames) { StringBuffer sb = new StringBuffer(); Pattern pattern = Pattern.compile("(\\w+)( *\\()"); Matcher matcher = pattern.matcher(sql); while (matcher.find()) { String displayName = matcher.group(1); String name = udfInternalNames.get(displayName); if (name != null && !name.equalsIgnoreCase(displayName)) { matcher.appendReplacement(sb, name + matcher.group(2)); } } matcher.appendTail(sb); return sb.toString(); } StreamCatalogService(StorageManager dao, FileStorage fileStorage, MLModelRegistryClient modelRegistryClient); Notifier addNotifierInfo(Notifier notifier); Notifier getNotifierInfo(Long id); Collection<Notifier> listNotifierInfos(); Collection<Notifier> listNotifierInfos(List<QueryParam> params); Notifier removeNotifierInfo(Long notifierId); Notifier addOrUpdateNotifierInfo(Long id, Notifier notifier); Collection<TopologyVersion> listCurrentTopologyVersionInfos(); Collection<TopologyVersion> listTopologyVersionInfos(List<QueryParam> queryParams); Optional<TopologyVersion> getCurrentTopologyVersionInfo(Long topologyId); Optional<TopologyVersion> getLatestVersionInfo(Long topologyId); TopologyVersion getTopologyVersionInfo(Long versionId); TopologyVersion addTopologyVersionInfo(TopologyVersion topologyVersion); TopologyVersion addOrUpdateTopologyVersionInfo(Long versionId, TopologyVersion topologyVersion); Long getVersionTimestamp(Long versionId); TopologyVersion updateVersionTimestamp(Long versionId); TopologyVersion updateVersionTimestamp(Long versionId, Long timestamp); TopologyVersion removeTopologyVersionInfo(Long versionId); Collection<Topology> listTopologies(); Collection<Topology> listTopologies(List<QueryParam> queryParams); Long getCurrentVersionId(Long topologyId); Topology getTopology(Long topologyId); Topology getTopology(Long topologyId, Long versionId); Topology addTopology(Topology topology); Topology removeTopology(Long topologyId, boolean recurse); Topology removeTopology(Long topologyId, Long versionId, boolean recurse); Topology cloneTopologyVersion(Long topologyId, Long versionId); Topology addOrUpdateTopology(Long topologyId, Topology topology); TopologyComponent getTopologyComponent(Long topologyId, Long topologyComponentId); String exportTopology(Topology topology); Topology importTopology(Long namespaceId, TopologyData topologyData); Topology cloneTopology(Long namespaceId, Topology topology); Optional<TopologyState> getTopologyState(Long topologyId); TopologyState addTopologyState(Long topologyId, TopologyState state); TopologyState addOrUpdateTopologyState(Long topologyID, TopologyState state); TopologyState removeTopologyState(Long topologyId); Collection<TopologyComponentBundle.TopologyComponentType> listTopologyComponentBundleTypes(); Collection<TopologyComponentBundle> listTopologyComponentBundlesForTypeWithFilter(TopologyComponentBundle.TopologyComponentType componentType,
                                                                                  List<QueryParam> params); TopologyComponentBundle getTopologyComponentBundle(Long topologyComponentBundleId); TopologyComponentBundle addTopologyComponentBundle(TopologyComponentBundle topologyComponentBundle, java.io.File bundleJar); TopologyComponentBundle addOrUpdateTopologyComponentBundle(Long id, TopologyComponentBundle topologyComponentBundle, java.io.File bundleJar); TopologyComponentBundle removeTopologyComponentBundle(Long id); InputStream getFileFromJarStorage(String fileName); Collection<CustomProcessorInfo> listCustomProcessorsFromBundleWithFilter(List<QueryParam> params); CustomProcessorInfo addCustomProcessorInfoAsBundle(CustomProcessorInfo customProcessorInfo, InputStream jarFile); CustomProcessorInfo updateCustomProcessorInfoAsBundle(CustomProcessorInfo customProcessorInfo, InputStream jarFile, boolean verify); Collection<CustomProcessorInfo> upgradeCustomProcessorsWithDigest(); String uploadFileToStorage(InputStream inputStream, String jarFileName); InputStream downloadFileFromStorage(String jarName); boolean deleteFileFromStorage(String jarName); CustomProcessorInfo removeCustomProcessorInfoAsBundle(String name); Optional<TopologyEditorToolbar> getTopologyEditorToolbar(long userId); Collection<TopologyEditorToolbar> listTopologyEditorToolbar(List<QueryParam> queryParams); TopologyEditorToolbar addTopologyEditorToolbar(TopologyEditorToolbar toolbar); TopologyEditorToolbar addOrUpdateTopologyEditorToolbar(TopologyEditorToolbar toolbar); Optional<TopologyEditorToolbar> removeTopologyEditorToolbar(long userId); Collection<TopologyEditorMetadata> listTopologyEditorMetadata(); Collection<TopologyEditorMetadata> listTopologyEditorMetadata(List<QueryParam> queryParams); TopologyEditorMetadata getTopologyEditorMetadata(Long topologyId); TopologyEditorMetadata getTopologyEditorMetadata(Long topologyId, Long versionId); TopologyEditorMetadata addTopologyEditorMetadata(Long topologyId,
                                                            TopologyEditorMetadata topologyEditorMetadata); TopologyEditorMetadata addTopologyEditorMetadata(Long topologyId,
                                                            Long versionId,
                                                            TopologyEditorMetadata topologyEditorMetadata); TopologyEditorMetadata addOrUpdateTopologyEditorMetadata(Long topologyId, TopologyEditorMetadata topologyEditorMetadata); TopologyEditorMetadata removeTopologyEditorMetadata(Long topologyId); TopologyEditorMetadata removeTopologyEditorMetadata(Long topologyId, Long versionId); TopologySource getTopologySource(Long topologyId, Long sourceId); TopologySource getTopologySource(Long topologyId, Long sourceId, Long versionId); TopologySource addTopologySource(Long topologyId, TopologySource topologySource); TopologySource addTopologySource(Long topologyId,
                                            Long versionId,
                                            TopologySource topologySource); Map<String, Set<Long>> getComponentsToReconfigure(Topology topology); TopologySource addOrUpdateTopologySource(Long topologyId, Long sourceId, TopologySource topologySource); TopologySource removeTopologySource(Long topologyId, Long sourceId, boolean removeEdges); TopologySource removeTopologySource(Long topologyId, Long sourceId, Long versionId, boolean removeEdges); Collection<TopologySource> listTopologySources(); Collection<TopologySource> listTopologySources(List<QueryParam> params); TopologySink getTopologySink(Long topologyId, Long sinkId); TopologySink getTopologySink(Long topologyId, Long sinkId, Long versionId); TopologySink addTopologySink(Long topologyId, TopologySink topologySink); TopologySink addTopologySink(Long topologyId,
                                        Long versionId,
                                        TopologySink topologySink); TopologySink addOrUpdateTopologySink(Long topologyId, Long id, TopologySink topologySink); TopologySink removeTopologySink(Long topologyId, Long sinkId, boolean removeEdges); TopologySink removeTopologySink(Long topologyId, Long sinkId, Long versionId, boolean removeEdges); Collection<TopologySink> listTopologySinks(); Collection<TopologySink> listTopologySinks(List<QueryParam> params); TopologyProcessor getTopologyProcessor(Long topologyId, Long processorId); TopologyProcessor getTopologyProcessor(Long topologyId, Long processorId, Long versionId); TopologyProcessor addTopologyProcessor(Long topologyId, TopologyProcessor topologyProcessor); TopologyProcessor addTopologyProcessor(Long topologyId,
                                                  Long versionId,
                                                  TopologyProcessor topologyProcessor); TopologyProcessor addOrUpdateTopologyProcessor(Long topologyId, Long id, TopologyProcessor topologyProcessor); TopologyProcessor removeTopologyProcessor(Long topologyId, Long processorId, boolean removeEdges); TopologyProcessor removeTopologyProcessor(Long topologyId, Long processorId, Long versionId, boolean removeEdges); Collection<TopologyProcessor> listTopologyProcessors(); Collection<TopologyProcessor> listTopologyProcessors(List<QueryParam> params); TopologyEdge getTopologyEdge(Long topologyId, Long edgeId); TopologyEdge getTopologyEdge(Long topologyId, Long edgeId, Long versionId); TopologyEdge addTopologyEdge(Long topologyId, TopologyEdge topologyEdge, boolean reconfigure); TopologyEdge addTopologyEdge(Long topologyId, TopologyEdge topologyEdge); TopologyEdge addTopologyEdge(Long topologyId,
                                        Long versionId,
                                        TopologyEdge topologyEdge); void setReconfigureOnAllComponentsInTopology(Topology topology); TopologyEdge addOrUpdateTopologyEdge(Long topologyId, Long id, TopologyEdge topologyEdge); TopologyEdge removeTopologyEdge(Long topologyId, Long edgeId); TopologyEdge removeTopologyEdge(Long topologyId, Long edgeId, Long versionId); Collection<TopologyEdge> listTopologyEdges(); Collection<TopologyEdge> listTopologyEdges(List<QueryParam> params); TopologyStream getStreamInfo(Long topologyId, Long streamId); TopologyStream getStreamInfo(Long topologyId, Long streamId, Long versionId); TopologyStream getStreamInfoByName(Long topologyId, String streamId); TopologyStream getStreamInfoByName(Long topologyId,
                                              String streamId,
                                              Long versionId); TopologyStream addStreamInfo(Long topologyId, TopologyStream topologyStream); TopologyStream addStreamInfo(Long topologyId,
                                        Long versionId,
                                        TopologyStream topologyStream); TopologyStream addOrUpdateStreamInfo(Long topologyId, Long id, TopologyStream stream); TopologyStream removeStreamInfo(Long topologyId, Long streamId); TopologyStream removeStreamInfo(Long topologyId, Long streamId, Long versionId); Collection<TopologyStream> listStreamInfos(); Collection<TopologyStream> listStreamInfos(List<QueryParam> params); Collection<TopologyRule> listRules(); Collection<TopologyRule> listRules(List<QueryParam> params); TopologyRule addRule(Long topologyId, TopologyRule topologyRule); TopologyRule addRule(Long topologyId,
                                Long versionId,
                                TopologyRule topologyRule); TopologyRule getRule(Long topologyId, Long ruleId); TopologyRule getRule(Long topologyId, Long ruleId, Long versionId); TopologyRule addOrUpdateRule(Long topologyId, Long ruleId, TopologyRule topologyRule); TopologyRule removeRule(Long topologyId, Long ruleId); TopologyRule removeRule(Long topologyId, Long ruleId, Long versionId); Collection<TopologyWindow> listWindows(); Collection<TopologyWindow> listWindows(List<QueryParam> params); Collection<TopologyBranchRule> listBranchRules(); Collection<TopologyBranchRule> listBranchRules(List<QueryParam> params); TopologyBranchRule addBranchRule(Long topologyId, TopologyBranchRule topologyBranchRule); TopologyBranchRule addBranchRule(Long topologyId,
                                            Long versionId,
                                            TopologyBranchRule topologyBranchRule); TopologyBranchRule getBranchRule(Long topologyId, Long ruleId); TopologyBranchRule getBranchRule(Long topologyId, Long ruleId, Long versionId); TopologyBranchRule addOrUpdateBranchRule(Long topologyId, Long ruleId, TopologyBranchRule topologyBranchRule); TopologyBranchRule removeBranchRule(Long topologyId, Long id); TopologyBranchRule removeBranchRule(Long topologyId, Long id, Long versionId); TopologyWindow addWindow(Long topologyId, TopologyWindow topologyWindow); TopologyWindow addWindow(Long topologyId,
                                    Long versionId,
                                    TopologyWindow topologyWindow); TopologyWindow getWindow(Long topologyId, Long windowId); TopologyWindow getWindow(Long topologyId, Long windowId, Long versionId); TopologyWindow addOrUpdateWindow(Long topologyId, Long windowId, TopologyWindow topologyWindow); TopologyWindow removeWindow(Long topologyId, Long windowId); TopologyWindow removeWindow(Long topologyId, Long windowId, Long versionId); Collection<UDF> listUDFs(); Collection<UDF> listUDFs(List<QueryParam> queryParams); UDF getUDF(Long id); UDF addUDF(UDF udf); static Map<String, Class<?>> loadUdfsFromJar(java.io.File jarFile); UDF removeUDF(Long id); UDF addOrUpdateUDF(Long udfId, UDF udf); Collection<TopologyTestRunHistory> listTopologyTestRunHistory(Long topologyId); Collection<TopologyTestRunHistory> listTopologyTestRunHistory(Long topologyId, Long versionId); TopologyTestRunHistory getTopologyTestRunHistory(Long id); TopologyTestRunHistory addTopologyTestRunHistory(TopologyTestRunHistory history); TopologyTestRunHistory addOrUpdateTopologyTestRunHistory(Long id, TopologyTestRunHistory history); TopologyTestRunHistory removeTopologyTestRunHistory(Long id); Collection<TopologyTestRunCase> listTopologyTestRunCase(Long topologyId, Long versionId); Collection<TopologyTestRunCase> listTopologyTestRunCase(List<QueryParam> queryParams); TopologyTestRunCase getTopologyTestRunCase(Long topologyId, Long testcaseId); TopologyTestRunCase addTopologyTestRunCase(TopologyTestRunCase testCase); TopologyTestRunCase addOrUpdateTopologyTestRunCase(Long id, TopologyTestRunCase testCase); TopologyTestRunCase removeTestRunCase(Long topologyId, Long testcaseId); TopologyTestRunCaseSource getTopologyTestRunCaseSourceBySourceId(Long testCaseId, Long sourceId); TopologyTestRunCaseSource getTopologyTestRunCaseSource(Long testcaseId, Long id); TopologyTestRunCaseSource addTopologyTestRunCaseSource(TopologyTestRunCaseSource testCaseSource); TopologyTestRunCaseSource addOrUpdateTopologyTestRunCaseSource(Long id, TopologyTestRunCaseSource testCaseSource); TopologyTestRunCaseSource removeTopologyTestRunCaseSourceBySourceId(Long testcaseId, Long sourceId); TopologyTestRunCaseSource removeTopologyTestRunCaseSource(Long id); Collection<TopologyTestRunCaseSource> listTopologyTestRunCaseSource(Long testCaseId); Collection<TopologyTestRunCaseSource> listTopologyTestRunCaseSource(List<QueryParam> queryParams); TopologyTestRunCaseSink getTopologyTestRunCaseSinkBySinkId(Long testCaseId, Long sinkId); TopologyTestRunCaseSink getTopologyTestRunCaseSink(Long testcaseId, Long id); TopologyTestRunCaseSink addTopologyTestRunCaseSink(TopologyTestRunCaseSink testCaseSink); TopologyTestRunCaseSink addOrUpdateTopologyTestRunCaseSink(Long id, TopologyTestRunCaseSink testCaseSink); TopologyTestRunCaseSink removeTopologyTestRunCaseSinkBySinkId(Long testcaseId, Long sinkId); TopologyTestRunCaseSink removeTopologyTestRunCaseSink(Long id); Collection<TopologyTestRunCaseSink> listTopologyTestRunCaseSink(Long testCaseId); Collection<TopologyTestRunCaseSink> listTopologyTestRunCaseSink(List<QueryParam> queryParams); static final long PLACEHOLDER_ID; }
@Test public void testtranslateFunctionsProjection() { String result; Map<String, String> table = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); table.put("FOO", "FOO_FN"); table.put("foo2", "foo2_fn"); result = streamCatalogService.translateFunctions("select foo(arg1), FOO(arg2), FOO (arg3) from stream", table); Assert.assertEquals("select FOO_FN(arg1), FOO_FN(arg2), FOO_FN (arg3) from stream", result); result = streamCatalogService.translateFunctions("select foo2(arg1, arg2), bar(arg2), BAZ(arg3, arg4) from stream", table); Assert.assertEquals("select foo2_fn(arg1, arg2), bar(arg2), BAZ(arg3, arg4) from stream", result); }
@Test public void testtranslateFunctionsCondition() { String result; Map<String, String> table = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); table.put("FOO", "FOO_FN"); table.put("foo2", "foo2_fn"); table.put("UPPER", "UPPER"); result = streamCatalogService.translateFunctions("select foo, foo1, foo2 from stream where FOO(a, b) = 'c'", table); Assert.assertEquals("select foo, foo1, foo2 from stream where FOO_FN(a, b) = 'c'", result); result = streamCatalogService.translateFunctions("select * from stream where FOO(a) = 'b' AND UPPER(a) = 'C'", table); Assert.assertEquals("select * from stream where FOO_FN(a) = 'b' AND UPPER(a) = 'C'", result); }