import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
@Override
public void createAlertEntry(AlertEntry alertEntry) {
    // get the alert config if any
    final Optional<AlertConfigEntry> configForServiceIdCodeIdCount =
            alertsConfigStore.getConfigForServiceIdCodeIdCount(alertEntry.getServiceId(), alertEntry.getErrorCode());
    // get the max count of alerts before sending mail
    final int maxCount = configForServiceIdCodeIdCount.isPresent() ?
            configForServiceIdCodeIdCount.get().getMaxCount() : 1;
    final String mailTemplate = configForServiceIdCodeIdCount.isPresent() ?
            configForServiceIdCodeIdCount.get().getMailTemplate() : "ticket";
    // define the expiry of the entry in the cache
    final IgniteCache<String, AlertEntry> alertsCache = getAlertsCache();
    // insert into the key value store
    alertsCache.put(alertEntry.getAlertId(), alertEntry);
    // send the mail notification if max is there
    final SqlFieldsQuery sql = new SqlFieldsQuery("select count(*) from AlertEntry where serviceId = '" + alertEntry.getServiceId() + "' and errorCode = '" + alertEntry.getErrorCode() + "'");
    final List<List<?>> count = alertsCache.query(sql).getAll();
    if (count != null && !count.isEmpty()) {
        final Long result = (Long) count.get(0).get(0);
        if (result >= maxCount) {
            logger.debug("max alerts count is reached for : {}, start sending mail alert {}", alertEntry.toString());
            sendMail(alertEntry, configForServiceIdCodeIdCount.isPresent() ? configForServiceIdCodeIdCount.get().getEmails() : Collections.emptyList(), mailTemplate);
        }
    }
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
@Override
public Future<Void> doAsyncDeleteMessagesTo(String persistenceId, long toSequenceNr) {
    return storage.execute(persistenceId, cache, (entityIdParam, cacheParam) -> {
        if (log.isDebugEnabled()) {
            log.debug("doAsyncDeleteMessagesTo persistenceId'{}' toSequenceNr : {}", persistenceId, toSequenceNr);
        }

        List<List<?>> seq = cache
                .query(new SqlFieldsQuery("select sequenceNr from JournalItem where sequenceNr <= ? and persistenceId=?")
                        .setArgs(toSequenceNr, persistenceId))
                .getAll();
        Set<Long> keys = listsToStreamLong(seq).collect(Collectors.toSet());

        if (log.isDebugEnabled()) {
            log.debug("remove keys {}", keys);
        }

        cache.removeAll(keys);
        return null;
    });
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
@Override
public Future<Void> doDeleteAsync(String persistenceId, SnapshotSelectionCriteria criteria) {
    return storage.execute(persistenceId, cache, (entityIdParam, cacheParam) -> {
        if (log.isDebugEnabled()) {
            log.debug("doDeleteAsync '{}' ({}; {})", persistenceId, criteria.minSequenceNr(), criteria.maxSequenceNr());
        }

        List<List<?>> seq = cache
                .query(new SqlFieldsQuery("select sequenceNr from SnapshotItem where sequenceNr >= ? AND sequenceNr <= ? AND timestamp >= ? AND timestamp <= ? and persistenceId=?")
                        .setArgs(criteria.minSequenceNr(), criteria.maxSequenceNr(), criteria.minTimestamp(), criteria.maxTimestamp(), persistenceId))
                .getAll();
        Set<Long> keys = listsToSetLong(seq);

        if (log.isDebugEnabled()) {
            log.debug("remove keys {}", keys);
        }
        cache.removeAll(keys);
        return null;
    });
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
private Iterable<List<?>> executeWithHints(IgniteCache<Object, BinaryObject> cache, SqlFieldsQuery sqlQuery, QueryHints hints) {
	Iterable<List<?>> result;

	if ( hints.isLocal() ) {
		if ( !provider.isClientMode() ) {
			sqlQuery.setLocal( true );
		}
	}
	if ( hints.isAffinityRun() ) {
		result = provider.affinityCall( cache.getName(), hints.getAffinityKey(), sqlQuery );
	}
	else {
		result = cache.query( sqlQuery );
	}

	return result;
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
@Override
public long getNumberOfAssociations(SessionFactory sessionFactory) {
	int associationCount = 0;
	IgniteDatastoreProvider datastoreProvider = getProvider( sessionFactory );
	for ( CollectionPersister collectionPersister : ( (SessionFactoryImplementor) sessionFactory ).getCollectionPersisters().values() ) {
		AssociationKeyMetadata associationKeyMetadata = ( (OgmCollectionPersister) collectionPersister ).getAssociationKeyMetadata();
		if ( associationKeyMetadata.getAssociationKind() == AssociationKind.ASSOCIATION ) {
			IgniteCache<Object, BinaryObject> associationCache = getAssociationCache( sessionFactory, associationKeyMetadata );
			StringBuilder query = new StringBuilder( "SELECT " )
										.append( StringHelper.realColumnName( associationKeyMetadata.getColumnNames()[0] ) )
										.append( " FROM " ).append( associationKeyMetadata.getTable() );
			SqlFieldsQuery sqlQuery = datastoreProvider.createSqlFieldsQueryWithLog( query.toString(), null );
			Iterable<List<?>> queryResult = associationCache.query( sqlQuery );
			Set<Object> uniqs = new HashSet<>();
			for ( List<?> row : queryResult ) {
				Object value = row.get( 0 );
				if ( value != null ) {
					uniqs.add( value );
				}
			}
			associationCount += uniqs.size();
		}
	}
	return associationCount;
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
private static void getMostPopulatedCities(IgniteCache countryCache) {
    SqlFieldsQuery query = new SqlFieldsQuery(
        "SELECT name, population FROM country " +
        "ORDER BY population DESC LIMIT 10");

    FieldsQueryCursor<List<?>> cursor = countryCache.query(query);

    Iterator<List<?>> iterator = cursor.iterator();

    System.out.println();
    System.out.println(">>> 10 Most Populated Cities:");

    while (iterator.hasNext()) {
        List row = iterator.next();

        System.out.println("    >>> " + row.get(1) + " people live in " + row.get(0));
    }
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
private static void getTopCitiesInThreeCountries(IgniteCache countryCache) {
    SqlFieldsQuery query = new SqlFieldsQuery(
        "SELECT country.name, city.name, MAX(city.population) as max_pop FROM country " +
        "JOIN city ON city.countrycode = country.code " +
        "WHERE country.code IN ('USA','RUS','CHN') " +
        "GROUP BY country.name, city.name ORDER BY max_pop DESC LIMIT 3");

    FieldsQueryCursor<List<?>> cursor = countryCache.query(query);

    Iterator<List<?>> iterator = cursor.iterator();

    System.out.println();
    System.out.println(">>> 3 Most Populated Cities in US, RUS and CHN:");

    while (iterator.hasNext()) {
        List row = iterator.next();

        System.out.println("    >>> " + row.get(2) + " people live in " + row.get(1) + ", " + row.get(0));
    }
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 * Example for SQL-based fields queries that return only required fields instead of whole key-value pairs.
 */
private static void sqlFieldsQueryWithJoin() {
    IgniteCache<?, ?> cache = Ignition.ignite().cache(EMPLOYEE_CACHE_NAME);

    // Create query to get names of all employees.
    SqlFieldsQuery qry = new SqlFieldsQuery(
        "select e.ename, d.dname " +
            "from Employee e, \"" + DEPARTMENT_CACHE_NAME + "\".Department d " +
            "where e.deptno = d.deptno");

    // Execute query to get collection of rows. In this particular
    // case each row will have one element with full name of an employees.
    Collection<List<?>> res = cache.query(qry).getAll();

    // Print persons' names and departments' names.
    logDecorated("==Names of all employees and departments they belong to (SQL join)==", res);
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 * Example for SQL-based fields queries that return only required fields instead of whole key-value pairs.
 */
private static void aggregateQuery() {
    IgniteCache<?, ?> cache = Ignition.ignite().cache(EMPLOYEE_CACHE_NAME);

    // Create query to get sum of salaries and number of summed rows.
    SqlFieldsQuery qry = new SqlFieldsQuery("select sum(sal), count(sal) from Employee");

    // Execute query to get collection of rows.
    Collection<List<?>> res = cache.query(qry).getAll();

    double sum = 0;
    long cnt = 0;

    for (List<?> row : res) {
        // Skip results from nodes without data.
        if (row.get(0) != null) {
            sum += ((BigDecimal)row.get(0)).doubleValue();
            cnt += (Long)row.get(1);
        }
    }

    // Average employee salary
    log("==Average employee salary (aggregation query)==");
    log("\t" + (cnt > 0 ? (sum / cnt) : "n/a"));
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 * Example for SQL-based fields queries that return only required fields instead of whole key-value pairs.
 */
private static void groupByQuery() {
    IgniteCache<?, ?> cache = Ignition.ignite().cache(EMPLOYEE_CACHE_NAME);

    // Create query to get salary averages grouped by department name.
    // We don't need to perform any extra manual steps here, because
    // Employee data is colocated based on department IDs.
    SqlFieldsQuery qry = new SqlFieldsQuery(
        "select avg(e.sal), d.dname " +
            "from Employee e, \"" + DEPARTMENT_CACHE_NAME + "\".Department d " +
            "where e.deptno = d.deptno " +
            "group by d.dname " +
            "having avg(e.sal) > ?");

    // Execute query to get collection of rows.
    logDecorated("==Average salaries per Department (group-by query)==", cache.query(qry.setArgs(500)).getAll());
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    // Mark this cluster member as client.
    Ignition.setClientMode(true);

    try (Ignite ignite = Ignition.start("example-ignite.xml")) {
        if (!ExamplesUtils.hasServerNodes(ignite))
            return;

        CacheConfiguration<String, Alert> alert_Cfg = new CacheConfiguration<>("alerts");
        IgniteCache<String, Alert> instCache = ignite.getOrCreateCache(alert_Cfg);

        SqlFieldsQuery top3qry = new SqlFieldsQuery(QUERY_RED);
        while(true){
            // Execute queries.
            List<List<?>> top3 = instCache.query(top3qry).getAll();

            System.out.println("Service Health Monitoring");

            ExamplesUtils.printQueryResults(top3);

            Thread.sleep(1000);
        }

    }
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    // Mark this cluster member as client.
    Ignition.setClientMode(true);

    try (Ignite ignite = Ignition.start("example-ignite.xml")) {
        if (!ExamplesUtils.hasServerNodes(ignite))
            return;
        // query code goes here.
        CacheConfiguration<String, ServiceStatus> healthchecksCfg = new CacheConfiguration<>("healthchecks");
        IgniteCache<String, ServiceStatus> instCache = ignite.getOrCreateCache(healthchecksCfg);
        SqlFieldsQuery query = new SqlFieldsQuery(QUERY_404);

        while(true){
            // Execute queries.
            List<List<?>> res = instCache.query(query).getAll();

            System.out.println("Service Health check status");

            ExamplesUtils.printQueryResults(res);

            Thread.sleep(1000);
        }

    }
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
public <L extends Serializable> Observable<L> getByQueryAsValue(Class<L> l, String query, Object[] params) {

        return Observable.create(observer -> {
            try {

                SqlFieldsQuery sql = new SqlFieldsQuery(query);


                // Execute the query and obtain the query result cursor.
                try (QueryCursor<List<?>> queryResult = getDatastoreCache().query(sql.setArgs(params))) {
                    // callback with value

                    for (List entry : queryResult) {
                        // callback with value
                        observer.onNext((L) entry.get(0));
                    }

                }

                observer.onCompleted();
            } catch (Exception e) {
                observer.onError(e);
            }
        });

    }
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public List<FieldsQueryCursor<List<?>>> queryMultipleStatements(SqlFieldsQuery qry) {
    A.notNull(qry, "qry");
    try {
        ctx.checkSecurity(SecurityPermission.CACHE_READ);

        validate(qry);

        convertToBinary(qry);

        CacheOperationContext opCtxCall = ctx.operationContextPerCall();

        boolean keepBinary = opCtxCall != null && opCtxCall.isKeepBinary();

        return ctx.kernalContext().query().querySqlFields(ctx, qry, keepBinary, false);
    }
    catch (Exception e) {
        if (e instanceof CacheException)
            throw (CacheException)e;

        throw new CacheException(e);
    }
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 *
 */
public void testMergeFromSubquery() {
    IgniteCache p = ignite(0).cache("S2P").withKeepBinary();

    p.query(new SqlFieldsQuery("merge into String (_key, _val) values ('s', ?), " +
        "('a', ?)").setArgs("Sergi", "Alex").setLocal(true));

    assertEquals("Sergi", p.get("s"));
    assertEquals("Alex", p.get("a"));

    p.query(new SqlFieldsQuery("merge into Person(_key, id, firstName) " +
        "(select substring(lower(_val), 0, 2), cast(length(_val) as int), _val from String)"));

    assertEquals(createPerson(5, "Sergi"), p.get("se"));

    assertEquals(createPerson(4, "Alex"), p.get("al"));
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 *
 */
public void testNestedFieldsHandling() {
    IgniteCache<Integer, AllTypes> p = ignite(0).cache("I2AT");

    p.query(new SqlFieldsQuery(
        "insert into AllTypes(_key, innerTypeCol, arrListCol, _val, innerStrCol) values (1, ?, ?, ?, 'sss')").
        setArgs(
            new AllTypes.InnerType(50L),
            new ArrayList<>(Arrays.asList(3L, 2L, 1L)),
            new AllTypes(1L)
        )
    );

    AllTypes res = p.get(1);

    AllTypes.InnerType resInner = new AllTypes.InnerType(50L);

    resInner.innerStrCol = "sss";
    resInner.arrListCol = new ArrayList<>(Arrays.asList(3L, 2L, 1L));

    assertEquals(resInner, res.innerTypeCol);
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 * Tests insertion of an entity.
 *
 * @throws Exception If failed.
 */
public void testInsertEntityFields() throws Exception {
    cache.remove(entity.getId());

    assertEquals(0, cache.size());

    SqlFieldsQuery qry = new SqlFieldsQuery(
        "insert into EntityWithJsr310Fields(_key, id, locTime, locDate, locDateTime) values(?, ?, ?, ?, ?)"
    ).setArgs(
        entity.getId(), entity.getId(), entity.getLocalTime(), entity.getLocalDate(), entity.getLocalDateTime()
    );

    List<List<?>> qryResults = cache.query(qry).getAll();

    assertEquals(1, qryResults.size());
    assertEquals(1L, qryResults.get(0).get(0));
    assertEquals(1, cache.size());
    assertEquals(entity, cache.get(entity.getId()));
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 *
 */
@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
public void testDuplicateKeysException() {
    final IgniteCache<Integer, Integer> p = ignite(0).cache("I2I");

    p.clear();

    p.put(3, 5);

    GridTestUtils.assertThrows(log, new Callable<Void>() {
        /** {@inheritDoc} */
        @Override public Void call() throws Exception {
            p.query(new SqlFieldsQuery("insert into Integer(_key, _val) values (1, ?), " +
                "(?, 4), (5, 6)").setArgs(2, 3));

            return null;
        }
    }, CacheException.class, "Failed to INSERT some keys because they are already in cache [keys=[3]]");

    assertEquals(2, (int)p.get(1));
    assertEquals(5, (int)p.get(3));
    assertEquals(6, (int)p.get(5));
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 * Test that after index creation index is used by queries.
 * @throws SQLException If failed.
 */
public void testCreateIndex() throws SQLException {
    assertSize(3);

    assertColumnValues(30, 20, 10);

    jdbcRun(CREATE_INDEX);

    // Test that local queries on all server nodes use new index.
    for (int i = 0 ; i < 3; i++) {
        List<List<?>> locRes = ignite(i).cache(DEFAULT_CACHE_NAME).query(new SqlFieldsQuery("explain select id from " +
            "Person where id = 5").setLocal(true)).getAll();

        assertEquals(F.asList(
            Collections.singletonList("SELECT\n" +
                "    ID\n" +
                "FROM \"" + DEFAULT_CACHE_NAME + "\".PERSON\n" +
                "    /* \"" + DEFAULT_CACHE_NAME + "\".IDX: ID = 5 */\n" +
                "WHERE ID = 5")
        ), locRes);
    }

    assertSize(3);

    assertColumnValues(30, 20, 10);
}
 

import org.apache.ignite.cache.query.SqlFieldsQuery; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testSqlQuery() throws Exception {
    Ignite node = ignite(0);

    IgniteCache c1 = node.createCache(personCacheConfiguration(GROUP1, "c1"));
    IgniteCache c2 = node.createCache(personCacheConfiguration(GROUP1, "c2"));

    SqlFieldsQuery qry = new SqlFieldsQuery("select name from Person where name=?");
    qry.setArgs("p1");

    assertEquals(0, c1.query(qry).getAll().size());
    assertEquals(0, c2.query(qry).getAll().size());

    c1.put(1, new Person("p1"));

    assertEquals(1, c1.query(qry).getAll().size());
    assertEquals(0, c2.query(qry).getAll().size());

    c2.put(2, new Person("p1"));

    assertEquals(1, c1.query(qry).getAll().size());
    assertEquals(1, c2.query(qry).getAll().size());
}