()
| 1854 | } |
| 1855 | |
| 1856 | private void postProcess() throws IOException { |
| 1857 | // The following code deals with store/load combination of |
| 1858 | // intermediate files. In this case we will replace the load |
| 1859 | // operator |
| 1860 | // with a (implicit) split operator, iff the load/store |
| 1861 | // func is reversible (because that's when we can safely |
| 1862 | // skip the load and keep going with the split output). If |
| 1863 | // the load/store func is not reversible (or they are |
| 1864 | // different functions), we connect the store and the load |
| 1865 | // to remember the dependency. |
| 1866 | |
| 1867 | Set<LOLoad> loadOps = new HashSet<LOLoad>(); |
| 1868 | List<Operator> sources = lp.getSources(); |
| 1869 | for (Operator source : sources) { |
| 1870 | if (source instanceof LOLoad) { |
| 1871 | loadOps.add((LOLoad)source); |
| 1872 | } |
| 1873 | } |
| 1874 | |
| 1875 | Set<LOStore> storeOps = new HashSet<LOStore>(); |
| 1876 | List<Operator> sinks = lp.getSinks(); |
| 1877 | for (Operator sink : sinks) { |
| 1878 | if (sink instanceof LOStore) { |
| 1879 | storeOps.add((LOStore)sink); |
| 1880 | } |
| 1881 | } |
| 1882 | |
| 1883 | if ("true".equals(pigContext.getProperties().getProperty(PIG_LOCATION_CHECK_STRICT))) { |
| 1884 | log.info("Output location strick check enabled"); |
| 1885 | checkDuplicateStoreLoc(storeOps); |
| 1886 | } |
| 1887 | |
| 1888 | for (LOLoad load : loadOps) { |
| 1889 | for (LOStore store : storeOps) { |
| 1890 | String ifile = load.getFileSpec().getFileName(); |
| 1891 | String ofile = store.getFileSpec().getFileName(); |
| 1892 | if (ofile.equals(ifile)) { |
| 1893 | // if there is no path from the load to the store, |
| 1894 | // then connect the store to the load to create the |
| 1895 | // dependency of the store on the load. If there is |
| 1896 | // a path from the load to the store, then we should |
| 1897 | // not connect the store to the load and create a cycle |
| 1898 | if (!store.getPlan().pathExists(load, store)) { |
| 1899 | store.getPlan().connect(store, load); |
| 1900 | } |
| 1901 | } |
| 1902 | } |
| 1903 | } |
| 1904 | } |
| 1905 | |
| 1906 | /** |
| 1907 | * This method checks whether the multiple sinks (STORE) use the same |
no test coverage detected